Exceptions

Advanced error handling and recovery system for APILinker.

This module provides sophisticated error handling capabilities including: 1. Circuit breakers to prevent cascading failures 2. Dead Letter Queue (DLQ) for failed operations 3. Configurable recovery strategies 4. Detailed error logging and analytics

exception apilinker.core.error_handling.ApiLinkerError(message: str, error_category: ErrorCategory = ErrorCategory.UNKNOWN, status_code: int | None = None, response_body: str | None = None, request_url: str | None = None, request_method: str | None = None, operation_id: str | None = None, correlation_id: str | None = None, additional_context: Dict[str, Any] | None = None)

Bases: Exception

Base exception class for APILinker errors with enhanced context.

classmethod from_exception(exc: Exception, **kwargs) → ApiLinkerError

Convert a standard exception to an ApiLinkerError with additional context.

to_dict() → Dict[str, Any]

Convert error to a dictionary for logging or serialization.

class apilinker.core.error_handling.CircuitBreaker(name: str, failure_threshold: int = 5, reset_timeout_seconds: float = 60.0, half_open_max_calls: int = 1)

Bases: Generic[T]

Circuit Breaker implementation to prevent cascading failures.

When a service is failing, the circuit breaker will “open” after a certain threshold of failures, preventing further calls to the failing service until a reset timeout has passed. This helps to:

1. Prevent overwhelming an already struggling service

2. Fail fast rather than waiting for timeouts

3. Allow the service time to recover

__init__(name: str, failure_threshold: int = 5, reset_timeout_seconds: float = 60.0, half_open_max_calls: int = 1)

Initialize a circuit breaker.

Parameters:

• name – Name of this circuit breaker for logging

• failure_threshold – Number of consecutive failures before opening circuit

• reset_timeout_seconds – Time to wait before trying again (moving to HALF_OPEN)

• half_open_max_calls – Number of test calls allowed in HALF_OPEN state

execute(operation: Callable[[], T]) → Tuple[T | None, ApiLinkerError | None]

Execute an operation with circuit breaker protection.

Parameters:

operation – The function to execute

Returns:

Tuple of (result, error)

property state: CircuitBreakerState

Get the current state of the circuit breaker.

class apilinker.core.error_handling.CircuitBreakerState(value)

Bases: Enum

States for the circuit breaker pattern.

CLOSED = 'CLOSED'

HALF_OPEN = 'HALF_OPEN'

OPEN = 'OPEN'

class apilinker.core.error_handling.DeadLetterQueue(storage_dir: str | None = None)

Bases: object

Dead Letter Queue for storing failed operations for later analysis or retry.

This implementation stores failed operations as JSON files in a specified directory for durability and easy inspection.

__init__(storage_dir: str | None = None)

Initialize the Dead Letter Queue.

Parameters:

storage_dir – Directory to store DLQ items. If None, defaults to a ‘dlq’ subdirectory in the current working directory.

add_item(error: ApiLinkerError, payload: Any, metadata: Dict[str, Any] | None = None) → str

Add a failed item to the Dead Letter Queue.

Parameters:

• error – The error that caused the operation to fail

• payload – The data that was being processed when the failure occurred

• metadata – Additional metadata about the operation

Returns:

ID of the DLQ item for reference

get_items(error_category: ErrorCategory | None = None, since_timestamp: str | None = None, limit: int = 100) → List[Dict[str, Any]]

Retrieve items from the Dead Letter Queue with optional filtering.

Parameters:

• error_category – Filter items by error category

• since_timestamp – Only return items after this timestamp

• limit – Maximum number of items to return

Returns:

List of DLQ items matching the criteria

retry_item(item_id: str, operation: Callable[[Any], Any]) → Tuple[bool, Any | None, ApiLinkerError | None]

Retry a specific item from the Dead Letter Queue.

Parameters:

• item_id – ID of the item to retry

• operation – Function to call with the payload

Returns:

Tuple of (success, result, error)

class apilinker.core.error_handling.ErrorAnalytics(max_errors: int = 1000)

Bases: object

Collects and analyzes error data to provide insights on system health.

__init__(max_errors: int = 1000)

Initialize error analytics.

Parameters:

max_errors – Maximum number of errors to keep in memory

get_error_rate(category: ErrorCategory | None = None, minutes: int = 5) → float

Get the error rate for a category over the specified time period.

Parameters:

• category – Error category or None for all categories

• minutes – Time window in minutes

Returns:

Errors per minute

get_summary() → Dict[str, Any]

Get a summary of error statistics.

get_top_errors(limit: int = 5) → List[Dict[str, Any]]

Get the most frequent error categories.

record_error(error: ApiLinkerError) → None

Record an error for analysis.

class apilinker.core.error_handling.ErrorCategory(value)

Bases: Enum

Categories of errors for better handling and reporting.

AUTHENTICATION = 'AUTH'

CLIENT = 'CLIENT'

MAPPING = 'MAPPING'

NETWORK = 'NETWORK'

PLUGIN = 'PLUGIN'

RATE_LIMIT = 'RATE_LIMIT'

SERVER = 'SERVER'

TIMEOUT = 'TIMEOUT'

UNKNOWN = 'UNKNOWN'

VALIDATION = 'VALIDATION'

class apilinker.core.error_handling.ErrorRecoveryManager(dlq: DeadLetterQueue | None = None)

Bases: object

Manages error recovery strategies for different types of operations.

This class centralizes error recovery logic to make it configurable and consistent across the application.

__init__(dlq: DeadLetterQueue | None = None)

Initialize the error recovery manager.

Parameters:

dlq – Dead letter queue for storing failed operations

get_circuit_breaker(name: str) → CircuitBreaker

Get or create a circuit breaker for a specific operation.

Parameters:

name – Name of the circuit breaker

Returns:

CircuitBreaker instance

get_strategies(error_category: ErrorCategory, operation_type: str | None = None) → List[RecoveryStrategy]

Get recovery strategies for an error category and operation type.

Parameters:

• error_category – Type of error

• operation_type – Optional specific operation type

Returns:

List of recovery strategies

handle_error(error: ApiLinkerError, payload: Any, operation: Callable[[Any], Any], operation_type: str, max_retries: int = 3, retry_delay: float = 1.0, retry_backoff_factor: float = 2.0) → Tuple[bool, Any, ApiLinkerError | None]

Handle an error using appropriate recovery strategies.

Parameters:

• error – The error that occurred

• payload – Data being processed when the error occurred

• operation – Function to call for retry attempts

• operation_type – Type of operation for strategy selection

• max_retries – Maximum number of retry attempts

• retry_delay – Initial delay between retries in seconds

• retry_backoff_factor – Multiplicative factor for retry delay

Returns:

Tuple of (success, result, error)

set_strategy(error_category: ErrorCategory, strategies: List[RecoveryStrategy], operation_type: str | None = None) → None

Set recovery strategies for an error category.

Parameters:

• error_category – Type of error

• strategies – List of strategies to apply in order

• operation_type – Optional specific operation type, or None for default

class apilinker.core.error_handling.RecoveryStrategy(value)

Bases: Enum

Strategies for recovering from errors.

CIRCUIT_BREAKER = 'CIRCUIT_BREAKER'

EXPONENTIAL_BACKOFF = 'EXPONENTIAL_BACKOFF'

FAIL_FAST = 'FAIL_FAST'

FALLBACK = 'FALLBACK'

RETRY = 'RETRY'

SKIP = 'SKIP'

apilinker.core.error_handling.create_error_handler(dlq_dir: str | None = None) → Tuple[DeadLetterQueue, ErrorRecoveryManager, ErrorAnalytics]

Create a complete error handling system.

Parameters:

dlq_dir – Directory to store DLQ items

Returns:

Tuple of (DeadLetterQueue, ErrorRecoveryManager, ErrorAnalytics)