Exceptions

Custom exceptions for the Affinity API client.

All exceptions inherit from AffinityError for easy catching of all library errors.

AffinityError

Bases: Exception

Base exception for all Affinity API errors.

Source code in affinity/exceptions.py

class AffinityError(Exception):
    """Base exception for all Affinity API errors."""

    def __init__(
        self,
        message: str,
        *,
        status_code: int | None = None,
        response_body: Any | None = None,
        diagnostics: ErrorDiagnostics | None = None,
    ):
        super().__init__(message)
        self.message = message
        self.status_code = status_code
        self.response_body = response_body
        self.diagnostics = diagnostics

    def __str__(self) -> str:
        base = self.message
        if self.status_code:
            base = f"[{self.status_code}] {base}"
        if self.diagnostics:
            # Include method + url if both present, or just url if only url present
            if self.diagnostics.method and self.diagnostics.url:
                base = f"{base} ({self.diagnostics.method} {self.diagnostics.url})"
            elif self.diagnostics.url:
                base = f"{base} (url={self.diagnostics.url})"
            if self.diagnostics.request_id:
                base = f"{base} [request_id={self.diagnostics.request_id}]"
        return base

AuthenticationError

Bases: AffinityError

401 Unauthorized - Invalid or missing API key.

Your API key is invalid or was not provided.

Source code in affinity/exceptions.py

class AuthenticationError(AffinityError):
    """
    401 Unauthorized - Invalid or missing API key.

    Your API key is invalid or was not provided.
    """

    pass

AuthorizationError

Bases: AffinityError

403 Forbidden - Insufficient permissions.

You don't have permission to access this resource or perform this action. This can happen with: - Private lists you don't have access to - Admin-only actions - Resource-level permission restrictions

Source code in affinity/exceptions.py

class AuthorizationError(AffinityError):
    """
    403 Forbidden - Insufficient permissions.

    You don't have permission to access this resource or perform this action.
    This can happen with:
    - Private lists you don't have access to
    - Admin-only actions
    - Resource-level permission restrictions
    """

    pass

BetaEndpointDisabledError

Bases: UnsupportedOperationError

Attempted to call a beta endpoint without opt-in.

Source code in affinity/exceptions.py

class BetaEndpointDisabledError(UnsupportedOperationError):
    """Attempted to call a beta endpoint without opt-in."""

    pass

CompanyNotFoundError

Bases: EntityNotFoundError

Company with the specified ID was not found.

Source code in affinity/exceptions.py

class CompanyNotFoundError(EntityNotFoundError):
    """Company with the specified ID was not found."""

    def __init__(self, company_id: int, **kwargs: Any):
        super().__init__("Company", company_id, **kwargs)

ConfigurationError

Bases: AffinityError

Configuration error - missing or invalid client configuration.

Check that you've provided: - A valid API key - Correct base URLs (if customizing)

Source code in affinity/exceptions.py

class ConfigurationError(AffinityError):
    """
    Configuration error - missing or invalid client configuration.

    Check that you've provided:
    - A valid API key
    - Correct base URLs (if customizing)
    """

    pass

ConflictError

Bases: AffinityError

409 Conflict - Resource conflict.

The request conflicts with the current state of the resource. For example: - Trying to create a person with an email that already exists - Concurrent modification conflicts

Source code in affinity/exceptions.py

class ConflictError(AffinityError):
    """
    409 Conflict - Resource conflict.

    The request conflicts with the current state of the resource.
    For example:
    - Trying to create a person with an email that already exists
    - Concurrent modification conflicts
    """

    pass

DeprecationWarning

Bases: AffinityError

Feature is deprecated and may be removed.

Source code in affinity/exceptions.py

class DeprecationWarning(AffinityError):
    """
    Feature is deprecated and may be removed.
    """

    pass

EntityNotFoundError

Bases: NotFoundError

Specific entity not found.

Provides type-safe context about which entity type was not found.

Source code in affinity/exceptions.py

class EntityNotFoundError(NotFoundError):
    """
    Specific entity not found.

    Provides type-safe context about which entity type was not found.
    """

    def __init__(
        self,
        entity_type: str,
        entity_id: int | str,
        **kwargs: Any,
    ):
        message = f"{entity_type} with ID {entity_id} not found"
        super().__init__(message, **kwargs)
        self.entity_type = entity_type
        self.entity_id = entity_id

FieldNotFoundError

Bases: NotFoundError

Field with the specified ID was not found.

Source code in affinity/exceptions.py

class FieldNotFoundError(NotFoundError):
    """Field with the specified ID was not found."""

    pass

ListNotFoundError

Bases: NotFoundError

List with the specified ID was not found.

Source code in affinity/exceptions.py

class ListNotFoundError(NotFoundError):
    """List with the specified ID was not found."""

    pass

NetworkError

Bases: AffinityError

Network-level error.

Failed to connect to the Affinity API. Check your internet connection and firewall settings.

Source code in affinity/exceptions.py

class NetworkError(AffinityError):
    """
    Network-level error.

    Failed to connect to the Affinity API.
    Check your internet connection and firewall settings.
    """

    def __init__(
        self,
        message: str,
        *,
        diagnostics: ErrorDiagnostics | None = None,
    ):
        super().__init__(message, diagnostics=diagnostics)

NotFoundError

Bases: AffinityError

404 Not Found - Resource doesn't exist.

The requested resource (person, company, list, etc.) was not found. This could mean: - The ID is invalid - The resource was deleted - You don't have access to view it

Source code in affinity/exceptions.py

class NotFoundError(AffinityError):
    """
    404 Not Found - Resource doesn't exist.

    The requested resource (person, company, list, etc.) was not found.
    This could mean:
    - The ID is invalid
    - The resource was deleted
    - You don't have access to view it
    """

    pass

OpportunityNotFoundError

Bases: EntityNotFoundError

Opportunity with the specified ID was not found.

Source code in affinity/exceptions.py

class OpportunityNotFoundError(EntityNotFoundError):
    """Opportunity with the specified ID was not found."""

    def __init__(self, opportunity_id: int, **kwargs: Any):
        super().__init__("Opportunity", opportunity_id, **kwargs)

PersonNotFoundError

Bases: EntityNotFoundError

Person with the specified ID was not found.

Source code in affinity/exceptions.py

class PersonNotFoundError(EntityNotFoundError):
    """Person with the specified ID was not found."""

    def __init__(self, person_id: int, **kwargs: Any):
        super().__init__("Person", person_id, **kwargs)

PolicyError

Bases: AffinityError

Raised when a client policy blocks an attempted operation.

Source code in affinity/exceptions.py

class PolicyError(AffinityError):
    """Raised when a client policy blocks an attempted operation."""

    pass

RateLimitError

Bases: AffinityError

429 Too Many Requests - Rate limit exceeded.

You've exceeded the API rate limit. Wait before retrying. Check the response headers for rate limit info: - X-Ratelimit-Limit-User-Reset: Seconds until per-minute limit resets - X-Ratelimit-Limit-Org-Reset: Seconds until monthly limit resets

Source code in affinity/exceptions.py

class RateLimitError(AffinityError):
    """
    429 Too Many Requests - Rate limit exceeded.

    You've exceeded the API rate limit. Wait before retrying.
    Check the response headers for rate limit info:
    - X-Ratelimit-Limit-User-Reset: Seconds until per-minute limit resets
    - X-Ratelimit-Limit-Org-Reset: Seconds until monthly limit resets
    """

    def __init__(
        self,
        message: str,
        *,
        retry_after: int | None = None,
        status_code: int | None = None,
        response_body: Any | None = None,
        diagnostics: ErrorDiagnostics | None = None,
    ):
        super().__init__(
            message,
            status_code=status_code,
            response_body=response_body,
            diagnostics=diagnostics,
        )
        self.retry_after = retry_after

ServerError

Bases: AffinityError

500/503 Internal Server Error - Server-side problem.

Something went wrong on Affinity's servers. Try again later, and contact support if the problem persists.

Source code in affinity/exceptions.py

class ServerError(AffinityError):
    """
    500/503 Internal Server Error - Server-side problem.

    Something went wrong on Affinity's servers.
    Try again later, and contact support if the problem persists.
    """

    pass

TimeoutError

Bases: AffinityError

Request timeout.

The request took too long to complete. This could be due to: - Network issues - Large data sets - Server overload

Source code in affinity/exceptions.py

class TimeoutError(AffinityError):
    """
    Request timeout.

    The request took too long to complete.
    This could be due to:
    - Network issues
    - Large data sets
    - Server overload
    """

    def __init__(
        self,
        message: str,
        *,
        diagnostics: ErrorDiagnostics | None = None,
    ):
        super().__init__(message, diagnostics=diagnostics)

TooManyResultsError

Bases: AffinityError

Raised when .all() exceeds the limit.

The default limit is 100,000 items (approximately 100MB for typical Person objects). This protects against OOM errors when paginating large datasets.

To resolve: - Use .pages() for streaming iteration (memory-efficient) - Add filters to reduce result size - Pass limit=None to .all() if you're certain you need all results - Pass a custom limit=500_000 if you need more than the default

Source code in affinity/exceptions.py

class TooManyResultsError(AffinityError):
    """
    Raised when ``.all()`` exceeds the limit.

    The default limit is 100,000 items (approximately 100MB for typical Person objects).
    This protects against OOM errors when paginating large datasets.

    To resolve:
    - Use ``.pages()`` for streaming iteration (memory-efficient)
    - Add filters to reduce result size
    - Pass ``limit=None`` to ``.all()`` if you're certain you need all results
    - Pass a custom ``limit=500_000`` if you need more than the default
    """

    pass

UnsafeUrlError

Bases: AffinityError

SDK blocked following a server-provided URL.

Raised when SafeFollowUrl policy rejects a URL (scheme/host/userinfo/redirect).

Source code in affinity/exceptions.py

class UnsafeUrlError(AffinityError):
    """
    SDK blocked following a server-provided URL.

    Raised when SafeFollowUrl policy rejects a URL (scheme/host/userinfo/redirect).
    """

    def __init__(self, message: str, *, url: str | None = None):
        super().__init__(
            message,
            diagnostics=ErrorDiagnostics(url=url) if url else None,
        )
        self.url = url

UnsupportedOperationError

Bases: AffinityError

Operation not supported by the current API version.

Some operations are only available in V1 or V2.

Source code in affinity/exceptions.py

class UnsupportedOperationError(AffinityError):
    """
    Operation not supported by the current API version.

    Some operations are only available in V1 or V2.
    """

    pass

ValidationError

Bases: AffinityError

400/422 Bad Request/Unprocessable Entity - Invalid request data.

The request data is malformed or logically invalid. Check the error message for details about what's wrong.

Source code in affinity/exceptions.py

class ValidationError(AffinityError):
    """
    400/422 Bad Request/Unprocessable Entity - Invalid request data.

    The request data is malformed or logically invalid.
    Check the error message for details about what's wrong.
    """

    def __init__(
        self,
        message: str,
        *,
        param: str | None = None,
        status_code: int | None = None,
        response_body: Any | None = None,
        diagnostics: ErrorDiagnostics | None = None,
    ):
        super().__init__(
            message,
            status_code=status_code,
            response_body=response_body,
            diagnostics=diagnostics,
        )
        self.param = param

    def __str__(self) -> str:
        base = super().__str__()
        if self.param:
            return f"{base} (param: {self.param})"
        return base

VersionCompatibilityError

Bases: AffinityError

Response shape mismatch suggests API version incompatibility.

TR-015: Raised when the SDK detects response-shape mismatches that appear version-related. This typically means the API key's configured v2 Default API Version differs from what the SDK expects.

Guidance: 1. Check your API key's v2 Default API Version in the Affinity dashboard 2. Ensure it matches the expected_v2_version configured in the SDK 3. See: https://developer.affinity.co/#section/Getting-Started/Versioning

Source code in affinity/exceptions.py

class VersionCompatibilityError(AffinityError):
    """
    Response shape mismatch suggests API version incompatibility.

    TR-015: Raised when the SDK detects response-shape mismatches that
    appear version-related. This typically means the API key's configured
    v2 Default API Version differs from what the SDK expects.

    Guidance:
    1. Check your API key's v2 Default API Version in the Affinity dashboard
    2. Ensure it matches the expected_v2_version configured in the SDK
    3. See: https://developer.affinity.co/#section/Getting-Started/Versioning
    """

    def __init__(
        self,
        message: str,
        *,
        expected_version: str | None = None,
        parsing_error: str | None = None,
        status_code: int | None = None,
        response_body: Any | None = None,
        diagnostics: ErrorDiagnostics | None = None,
    ):
        super().__init__(
            message,
            status_code=status_code,
            response_body=response_body,
            diagnostics=diagnostics,
        )
        self.expected_version = expected_version
        self.parsing_error = parsing_error

    def __str__(self) -> str:
        base = super().__str__()
        hints = []
        if self.expected_version:
            hints.append(f"expected_v2_version={self.expected_version}")
        if self.parsing_error:
            hints.append(f"parsing_error={self.parsing_error}")
        if hints:
            base = f"{base} ({', '.join(hints)})"
        return base

WebhookInvalidJsonError

Bases: WebhookParseError

Raised when a webhook payload cannot be decoded as JSON.

Source code in affinity/exceptions.py

class WebhookInvalidJsonError(WebhookParseError):
    """Raised when a webhook payload cannot be decoded as JSON."""

    pass

WebhookInvalidPayloadError

Bases: WebhookParseError

Raised when a decoded webhook payload is not in the expected envelope shape.

Source code in affinity/exceptions.py

class WebhookInvalidPayloadError(WebhookParseError):
    """Raised when a decoded webhook payload is not in the expected envelope shape."""

    pass

WebhookInvalidSentAtError

Bases: WebhookParseError

Raised when a webhook sent_at field is missing or invalid.

Source code in affinity/exceptions.py

class WebhookInvalidSentAtError(WebhookParseError):
    """Raised when a webhook `sent_at` field is missing or invalid."""

    pass

WebhookMissingKeyError

Bases: WebhookParseError

Raised when a webhook payload is missing a required key.

Source code in affinity/exceptions.py

class WebhookMissingKeyError(WebhookParseError):
    """Raised when a webhook payload is missing a required key."""

    def __init__(self, message: str, *, key: str):
        super().__init__(message)
        self.key = key

WebhookParseError

Bases: AffinityError

Base error for inbound webhook parsing/validation failures.

Source code in affinity/exceptions.py

class WebhookParseError(AffinityError):
    """Base error for inbound webhook parsing/validation failures."""

    pass

WriteNotAllowedError

Bases: PolicyError

Raised when a write operation is attempted while the write policy denies writes.

Source code in affinity/exceptions.py

class WriteNotAllowedError(PolicyError):
    """Raised when a write operation is attempted while the write policy denies writes."""

    def __init__(self, message: str, *, method: str, url: str):
        super().__init__(message, diagnostics=ErrorDiagnostics(method=method, url=url))
        self.method = method
        self.url = url

error_from_response(status_code: int, response_body: Any, *, retry_after: int | None = None, diagnostics: ErrorDiagnostics | None = None) -> AffinityError

Create the appropriate exception from an API error response.

Parameters:

Name	Type	Description	Default
status_code	int	HTTP status code	required
response_body	Any	Parsed response body (usually dict with 'errors')	required
retry_after	int | None	Retry-After header value for rate limits	None

Returns:

Type	Description
AffinityError	Appropriate AffinityError subclass

Source code in affinity/exceptions.py

def error_from_response(
    status_code: int,
    response_body: Any,
    *,
    retry_after: int | None = None,
    diagnostics: ErrorDiagnostics | None = None,
) -> AffinityError:
    """
    Create the appropriate exception from an API error response.

    Args:
        status_code: HTTP status code
        response_body: Parsed response body (usually dict with 'errors')
        retry_after: Retry-After header value for rate limits

    Returns:
        Appropriate AffinityError subclass
    """
    # Try to extract message from response
    message = "Unknown error"
    param = None

    extracted = False
    if isinstance(response_body, dict):
        errors = response_body.get("errors")
        if isinstance(errors, list) and errors:
            for item in errors:
                if isinstance(item, dict):
                    msg = item.get("message")
                    if isinstance(msg, str) and msg.strip():
                        message = msg.strip()
                        p = item.get("param")
                        if isinstance(p, str) and p.strip():
                            param = p
                        extracted = True
                        break
                elif isinstance(item, str) and item.strip():
                    message = item.strip()
                    extracted = True
                    break

        if not extracted:
            top_message = response_body.get("message")
            if isinstance(top_message, str) and top_message.strip():
                message = top_message.strip()
                extracted = True
            else:
                detail = response_body.get("detail")
                if isinstance(detail, str) and detail.strip():
                    message = detail.strip()
                    extracted = True
                else:
                    error_obj = response_body.get("error")
                    if isinstance(error_obj, dict):
                        nested_message = error_obj.get("message")
                        if isinstance(nested_message, str) and nested_message.strip():
                            message = nested_message.strip()
                            extracted = True
                    elif isinstance(error_obj, str) and error_obj.strip():
                        message = error_obj.strip()
                        extracted = True

    if not extracted and isinstance(response_body, list) and response_body:
        first = response_body[0]
        if isinstance(first, dict):
            msg = first.get("message") or first.get("error") or first.get("detail")
            if isinstance(msg, str) and msg.strip():
                message = msg.strip()
                extracted = True
        elif isinstance(first, str) and first.strip():
            message = first.strip()
            extracted = True

    if (
        message == "Unknown error"
        and diagnostics is not None
        and isinstance(diagnostics.response_body_snippet, str)
    ):
        snippet = diagnostics.response_body_snippet.strip()
        if snippet and snippet not in {"{}", "[]"}:
            message = snippet

    # Map status codes to exceptions
    error_mapping: dict[int, type[AffinityError]] = {
        400: ValidationError,
        401: AuthenticationError,
        403: AuthorizationError,
        404: NotFoundError,
        409: ConflictError,
        422: ValidationError,
        429: RateLimitError,
        500: ServerError,
        502: ServerError,
        503: ServerError,
        504: ServerError,
    }

    error_class = error_mapping.get(status_code, AffinityError)

    # Special handling for ValidationError with param
    if error_class is ValidationError:
        return ValidationError(
            message,
            param=param,
            status_code=status_code,
            response_body=response_body,
            diagnostics=diagnostics,
        )

    # Special handling for RateLimitError with retry_after
    if error_class is RateLimitError:
        return RateLimitError(
            message,
            retry_after=retry_after,
            status_code=status_code,
            response_body=response_body,
            diagnostics=diagnostics,
        )

    return error_class(
        message,
        status_code=status_code,
        response_body=response_body,
        diagnostics=diagnostics,
    )