Skip to content

Error Handling Examples

This document provides examples of how to use the error handling system in different scenarios.

Basic Error Handling

from archipy.models.errors import (
    NotFoundError,
    InvalidArgumentError,
    PermissionDeniedError
)
from archipy.models.types.language_type import LanguageType

def get_user(user_id: str):
    try:
        # Attempt to fetch user
        user = user_repository.find_by_id(user_id)
        if not user:
            raise NotFoundError(
                resource_type="user",
                error_details=f"User with ID {user_id} not found",
                lang=LanguageType.EN
            )
        return user
    except NotFoundError as e:
        # Log the error
        logger.error(f"User not found: {e.to_dict()}")
        # Re-raise or handle as needed
        raise

def update_user_permissions(user_id: str, permissions: list[str]):
    try:
        # Validate input
        if not isinstance(permissions, list):
            raise InvalidArgumentError(
                argument="permissions",
                error_details="Permissions must be a list",
                lang=LanguageType.EN
            )

        # Check permissions
        if not has_admin_access():
            raise PermissionDeniedError(
                operation="update_permissions",
                error_details="Admin access required",
                lang=LanguageType.EN
            )

        # Update permissions
        user_repository.update_permissions(user_id, permissions)
    except (InvalidArgumentError, PermissionDeniedError) as e:
        logger.error(f"Failed to update permissions: {e.to_dict()}")
        raise

Business Logic Error Handling

from archipy.models.errors import (
    InsufficientFundsError,
    BusinessRuleViolationError,
    InvalidStateError
)

def process_transaction(account_id: str, amount: float):
    try:
        # Check account state
        account = account_repository.find_by_id(account_id)
        if not account.is_active:
            raise InvalidStateError(
                current_state="inactive",
                expected_state="active",
                error_details="Account must be active for transactions"
            )

        # Check balance
        if account.balance < amount:
            raise InsufficientFundsError(
                error_details=f"Required amount: {amount}, Available: {account.balance}"
            )

        # Check business rules
        if amount > account.transaction_limit:
            raise BusinessRuleViolationError(
                rule="transaction_limit",
                details=f"Amount exceeds limit of {account.transaction_limit}"
            )

        # Process transaction
        account_repository.process_transaction(account_id, amount)
    except (InsufficientFundsError, BusinessRuleViolationError, InvalidStateError) as e:
        logger.error(f"Transaction failed: {e.to_dict()}")
        raise

System Error Handling

from archipy.models.errors import (
    DatabaseConnectionError,
    DeadlockDetectedError,
    ResourceExhaustedError
)
from typing import Any

def execute_with_retry(operation: callable, max_retries: int = 3) -> Any:
    retries = 0
    while retries < max_retries:
        try:
            return operation()
        except DeadlockDetectedError as e:
            retries += 1
            if retries == max_retries:
                logger.error(f"Max retries exceeded: {e.to_dict()}")
                raise
            logger.warning(f"Deadlock detected, retrying ({retries}/{max_retries})")
            time.sleep(1)  # Wait before retry
        except DatabaseConnectionError as e:
            logger.error(f"Database connection failed: {e.to_dict()}")
            raise  # Don't retry connection errors
        except ResourceExhaustedError as e:
            logger.error(f"Resource exhausted: {e.to_dict()}")
            raise  # Don't retry resource exhaustion

# Usage example
def process_batch(items: list[dict]):
    def batch_operation():
        return database.batch_insert(items)

    try:
        return execute_with_retry(batch_operation)
    except (DeadlockDetectedError, DatabaseConnectionError, ResourceExhaustedError) as e:
        # Handle final failure
        return {"error": e.to_dict()}

Error Response Formatting

from fastapi import HTTPException
from fastapi.responses import JSONResponse
from archipy.models.errors import BaseError
from archipy.models.types.language_type import LanguageType
from datetime import datetime

def handle_error(error: BaseError) -> dict:
    """Convert error to API response format."""
    error_dict = error.to_dict()
    return {
        "status": "error",
        "error": error_dict,
        "timestamp": datetime.utcnow().isoformat()
    }

# FastAPI error handler
@app.exception_handler(BaseError)
async def error_handler(request: Request, exc: BaseError):
    return JSONResponse(
        status_code=exc.http_status_code or 500,
        content=handle_error(exc)
    )

# Example usage in endpoint
@app.get("/users/{user_id}")
async def get_user(user_id: str):
    try:
        user = user_service.get_user(user_id)
        return {"status": "success", "data": user}
    except NotFoundError as e:
        # Let the exception handler handle it
        raise

Error Logging and Monitoring

from archipy.models.errors import (
    InternalError,
    UnknownError,
    ConfigurationError
)
import sentry_sdk

def log_error(error: BaseError, context: dict | None = None):
    """Log error with context and send to monitoring service."""
    error_dict = error.to_dict()

    # Add context if provided
    if context:
        error_dict["context"] = context

    # Log to application logger
    logger.error(
        f"Error occurred: {error_dict['error']}",
        extra={"error": error_dict}
    )

    # Send to monitoring service
    if isinstance(error, (InternalError, UnknownError, ConfigurationError)):
        sentry_sdk.capture_exception(error)

# Example usage
def process_request(request_data: dict):
    try:
        # Process request
        result = service.process(request_data)
        return result
    except BaseError as e:
        log_error(e, context={"request_data": request_data})
        raise

Exception Chaining

from archipy.models.errors import (
    DatabaseQueryError,
    InvalidEntityTypeError,
    BaseEntity
)

# Good - Preserving original error context
def fetch_entity(entity_type: type, entity_uuid: str) -> BaseEntity:
    try:
        result = session.get(entity_type, entity_uuid)
        if not result:
            raise NotFoundError(
                resource_type=entity_type.__name__,
                error_details=f"Entity with UUID {entity_uuid} not found"
            )
        return result
    except Exception as e:
        raise DatabaseQueryError() from e

# Good - Type validation with specific error
def validate_entity(entity: object) -> None:
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__
        )

Error Recovery Strategies

from archipy.models.errors import (
    CacheMissError,
    ServiceUnavailableError,
    ResourceExhaustedError
)

class ErrorRecovery:
    @staticmethod
    def handle_cache_miss(error: CacheMissError):
        """Handle cache miss by fetching from primary source."""
        try:
            # Fetch from database
            data = database.get(error.key)
            # Update cache
            cache.set(error.key, data)
            return data
        except Exception as e:
            logger.error(f"Failed to recover from cache miss: {str(e)}")
            raise  # Re-raise after logging

    @staticmethod
    def handle_service_unavailable(error: ServiceUnavailableError):
        """Handle service unavailability with fallback."""
        if error.service == "primary":
            try:
            # Try fallback service
            return fallback_service.get_data()
            except Exception as e:
                # Preserve error chain
                raise ServiceUnavailableError(
                    service="fallback",
                    error_details="Both primary and fallback services unavailable"
                ) from e
        raise

    @staticmethod
    def handle_resource_exhaustion(error: ResourceExhaustedError):
        """Handle resource exhaustion with cleanup."""
        if error.resource_type == "memory":
            # Perform cleanup
            gc.collect()
            try:
            # Retry operation
            return retry_operation()
            except Exception as e:
                # Preserve error chain
                raise ResourceExhaustedError(
                    resource_type="memory",
                    error_details="Resource exhaustion persisted after cleanup"
                ) from e
        raise

# Example usage
def get_data(key: str):
    try:
        return cache.get(key)
    except CacheMissError as e:
        return ErrorRecovery.handle_cache_miss(e)
    except ServiceUnavailableError as e:
        return ErrorRecovery.handle_service_unavailable(e)
    except ResourceExhaustedError as e:
        return ErrorRecovery.handle_resource_exhaustion(e)