feat(superoffice): POC API handshake & auth flow [2ff88f42]

Establishes the initial structure for the SuperOffice connector. Implements the complete, iterative authentication process, culminating in a successful refresh token exchange. Documents the process and the final blocker (API authorization) in the integration plan, awaiting IT action to change the application type to 'Server to server'.
This commit is contained in:
2026-02-06 13:52:44 +00:00
parent 0f9e651d95
commit 53b92c76de
10 changed files with 310 additions and 24 deletions

View File

@@ -0,0 +1,55 @@
import requests
import logging
from .auth_handler import AuthHandler
logger = logging.getLogger(__name__)
class SuperOfficeClient:
"""
A client for interacting with the SuperOffice API.
"""
def __init__(self, auth_handler: AuthHandler, tenant_id: str):
# Base URL for the SuperOffice REST API, including tenant_id
self.base_url = f"https://sod.superoffice.com/{tenant_id}/api"
self.auth_handler = auth_handler
self.tenant_id = tenant_id
self.session = requests.Session()
def _get_auth_headers(self):
"""Returns the authorization headers with a valid token and context."""
access_token = self.auth_handler.get_access_token()
return {
'Authorization': f'Bearer {access_token}',
'Accept': 'application/json',
'X-SuperOffice-Context': f'TenantId={self.tenant_id}' # Crucial for multi-tenant environments
}
def test_connection(self):
"""
Performs a simple API call to test the connection and authentication.
Fetches the current user principal.
"""
endpoint = "/v1/User/currentPrincipal"
test_url = f"{self.base_url}{endpoint}"
logger.info(f"Attempting to test connection to: {test_url}")
try:
headers = self._get_auth_headers()
response = self.session.get(test_url, headers=headers)
response.raise_for_status()
user_data = response.json()
logger.info("Successfully connected to SuperOffice API.")
logger.info(f"Authenticated as: {user_data.get('Name', 'N/A')} ({user_data.get('Associate', 'N/A')})")
return user_data
except requests.exceptions.HTTPError as http_err:
logger.error(f"HTTP error during connection test: {http_err} - {http_err.response.text}")
return None
except requests.exceptions.RequestException as req_err:
logger.error(f"Request error during connection test: {req_err}")
return None
except Exception as e:
logger.error(f"An unexpected error occurred during connection test: {e}")
return None