Floke
ff62024ef7
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'.
56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
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
|