Standard Error Handling

Lumos offers a built-in exception handler that can be easily integrated into your connector:

# Example: abc/integration.py
from connector.oai.integration import Integration
from connector.oai.errors import HTTPHandler
from httpx import HTTPStatusError

integration = Integration(
    app_id="abc",
    auth=OAuthCredential,
    exception_handlers=[(HTTPStatusError, HTTPHandler, None)],
    # ... other configuration ...
)

This setup ensures that all HTTP error status codes are automatically handled in subsequent API calls. The HTTPHandler provides a consistent way to process and report HTTP errors across different connectors.

Custom Error Handling

For more specific error scenarios or to handle API-specific error codes, you can implement custom error handlers. Let's examine a sample connector's custom error handling as an example.

Parse Error Response

import json
import httpx
from abc.dto.error import abcErrorResponse

def parse_error_response_body(response: httpx.Response) -> abcErrorResponse:
    """Parse response body from abc response in error state."""
    try:
        json_data = response.json()
        return abcErrorResponse(**json_data)
    except json.JSONDecodeError:
        response.raise_for_status()
        # This next line shouldn't ever be hit - this function should only
        # be called for 4xx or 5xx errors, so raise_for_status() will raise
        return abcErrorResponse(code=-1, message="Error response that is unhandlable")

This function attempts to parse the JSON response from the API. If successful, it creates a abcErrorResponse object. If parsing fails, it raises the HTTP error or returns a default error response.

Define Custom Exceptions

# file at abc/errors.py
class abcError(Exception):
    def __init__(
        self,
        capability: CapabilityName,
        entitlement: str | abcEntitlementType | None,
        message: str,
        abc_code: int | None = None,
    ):
        self.message = message
        self.abc_code = abc_code
        self.capability = capability
        self.entitlement = entitlement

This custom exception allows you to include abc-specific error details like the error code, affected capability, and entitlement.

Implement Custom Error Handler

# file at abc/errors.py
from connector.oai.errors import ErrorCode, ExceptionHandler

class abcErrorHandler(ExceptionHandler):
    @staticmethod
    def handle(
        e: Exception,
        original_func: Any,
        response: ErrorResponse,
        error_code: str | ErrorCode | None = None,
    ) -> ErrorResponse:
        if isinstance(e, abcError):
            match e.abc_code:
                case 124:
                    response.error.message = e.message or "Invalid access token."
                    response.error.error_code = ErrorCode.UNAUTHORIZED
                    response.error.status_code = httpx.codes.UNAUTHORIZED
                case 200:
                    if (
                        e.capability == CapabilityName.ASSIGN_ENTITLEMENT
                        and e.entitlement == abcEntitlementType.CONTACT_GROUP
                    ):
                        response.error.message = "Only paid accounts can be added."
                        response.error.error_code = ErrorCode.BAD_REQUEST
                # ... more case statements for other error codes ...
        return response

This custom handler maps abc-specific error codes to appropriate Lumos error codes and messages, providing more context and better error reporting.

Register the Custom Error Handler

# file at abc/integration.py
from abc.errors import abcError, abcErrorHandler

integration = Integration(
    app_id="abc",
    auth=OAuthCredential,
    exception_handlers=[
        (HTTPStatusError, HTTPHandler, None),
        (abcError, abcErrorHandler, None)
    ],
    # ... other configuration ...
)

By adding the custom error handler to the exception_handlers list, you ensure that both standard HTTP errors and abc-specific errors are properly handled. You can now go ahead and raise your custom errors from within the connector.

# Example: abc/capabilities_write.py
# ...
response = await client.request(**request_kwargs, params=params)
if response.is_error:
    parsed_error = parse_error_response_body(response)
    raise AbcError(
        capability=StandardCapabilityName.UNASSIGN_ENTITLEMENT,
        entitlement=entitlement_type,
        message=parsed_error.message or "",
        abc_code=parsed_error.code or None,
    )
# ...

Best Practises

API-Specific Errors: Map API-specific error codes to meaningful messages and appropriate Lumos error codes.
Contextual Information: Include relevant context (e.g., capability, entitlement) in custom exceptions to aid in error diagnosis.
Consistent Error Codes: Map custom errors to standard Lumos ErrorCode values for consistency across connectors.
Fallback Handling: Always include a fallback case to handle unexpected errors gracefully.