Pattern-007: Async Error Handling Pattern

Status

Proven

Context

Asynchronous operations in modern applications introduce complex error handling challenges where failures can occur at multiple points in the execution pipeline. Traditional synchronous error handling patterns don’t adequately address async contexts where errors may be deferred, propagated across async boundaries, or lost in concurrent operations. The Async Error Handling Pattern addresses:

Pattern Description

The Async Error Handling Pattern provides systematic error management for asynchronous operations through structured exception hierarchies, graceful degradation mechanisms, and user-friendly error reporting.

Core Principle: “Errors should be user-friendly, actionable, and provide clear recovery guidance, even in async contexts.”

Error Handling Model:

Async Request → Processing → Error Detection → Classification → User Response → Recovery Guidance

Key Components:

Implementation

Structure

# Base error hierarchy for async operations
class APIError(Exception):
    """Base class for all application-specific API errors."""

    def __init__(self, status_code: int, error_code: str, details: Dict[str, Any] = None):
        self.status_code = status_code
        self.error_code = error_code
        self.details = details or {}
        super().__init__(f"API Error [{error_code}]")

class AsyncOperationError(APIError):
    """Base class for async operation failures."""

    def __init__(self, operation: str, details: Dict[str, Any] = None):
        super().__init__(500, f"ASYNC_{operation.upper()}_FAILED", details)
        self.operation = operation

class IntentClassificationFailedError(AsyncOperationError):
    """Raised when async intent classification fails."""

    def __init__(self, query: str, details: Dict[str, Any] = None):
        super().__init__("INTENT_CLASSIFICATION", {
            "query": query,
            "suggestion": "Try rephrasing your request more clearly",
            **(details or {})
        })

Code Example

import asyncio
import logging
from typing import Optional, Dict, Any

class AsyncErrorHandler:
    """Centralized async error handling with graceful degradation."""

    def __init__(self, logger: Optional[logging.Logger] = None):
        self.logger = logger or logging.getLogger(__name__)

    async def handle_async_operation(
        self,
        operation: Callable,
        fallback: Optional[Callable] = None,
        context: Dict[str, Any] = None
    ) -> Any:
        """Execute async operation with error handling and optional fallback."""

        try:
            result = await operation()
            self.logger.info(f"Async operation succeeded: {context}")
            return result

        except APIError as e:
            # Known API errors - provide structured response
            self.logger.error(f"API error in async operation: {e.error_code}",
                            extra={"context": context, "details": e.details})

            if fallback:
                try:
                    fallback_result = await fallback()
                    self.logger.info("Fallback operation succeeded")
                    return fallback_result
                except Exception as fallback_error:
                    self.logger.error(f"Fallback also failed: {fallback_error}")

            raise e

        except asyncio.TimeoutError:
            # Timeout handling with recovery guidance
            timeout_error = AsyncOperationError("TIMEOUT", {
                "context": context,
                "suggestion": "Try again in a few moments or check system status"
            })
            self.logger.error("Async operation timed out", extra={"context": context})
            raise timeout_error

        except Exception as e:
            # Unexpected errors - wrap in structured format
            unexpected_error = AsyncOperationError("UNEXPECTED", {
                "original_error": str(e),
                "context": context,
                "suggestion": "Please report this issue to support"
            })
            self.logger.error(f"Unexpected error in async operation: {e}",
                            extra={"context": context}, exc_info=True)
            raise unexpected_error

# Usage example
async def process_user_request_with_error_handling(request: Dict[str, Any]) -> Dict[str, Any]:
    """Process user request with comprehensive async error handling."""

    error_handler = AsyncErrorHandler()

    async def main_operation():
        # Main async processing logic
        intent = await classify_intent(request["query"])
        response = await generate_response(intent, request)
        return {"status": "success", "response": response}

    async def fallback_operation():
        # Fallback when main operation fails
        return {
            "status": "degraded",
            "response": "I'm having trouble processing your request. Please try again.",
            "recovery_actions": ["Rephrase your request", "Check system status", "Contact support"]
        }

    return await error_handler.handle_async_operation(
        operation=main_operation,
        fallback=fallback_operation,
        context={"request_id": request.get("id"), "user": request.get("user")}
    )

Configuration

# Async Error Handling Configuration
error_handling:
  async:
    timeout_seconds: 30
    max_retries: 3
    retry_delay_seconds: 1

  graceful_degradation:
    enabled: true
    fallback_timeout_seconds: 10

  logging:
    log_all_errors: true
    include_stack_traces: true
    log_recovery_attempts: true

  user_messages:
    include_recovery_guidance: true
    technical_details: false
    contact_support_threshold: 3

Usage Guidelines

When to Use

When NOT to Use

Best Practices

Examples in Codebase

Primary Usage

Test Examples

Complements

Alternatives

Dependencies

Migration Notes

Consolidated from multiple sources

From PATTERN-INDEX.md

From error-handling-framework.md

References

Documentation

Usage Analysis

Pattern extracted and consolidated: September 15, 2025 Agent B (Cursor) - Pattern Catalog Consolidation Project