error-handling-logging-patterns

Error Handling & Logging Patterns

Build systems that are debuggable in production through structured logging and intentional error handling.

Error Hierarchy Design

Custom Exception Classes

class AppError(Exception):
    """Base error for the application."""
    def __init__(self, message: str, code: str = "INTERNAL_ERROR", status: int = 500):
        self.message = message
        self.code = code
        self.status = status
        super().__init__(message)

class NotFoundError(AppError): def init(self, entity: str, id: str): super().init(f"{entity} '{id}' not found", code="NOT_FOUND", status=404)

class ValidationError(AppError): def init(self, field: str, reason: str): super().init(f"Invalid {field}: {reason}", code="VALIDATION_ERROR", status=400)

class ExternalServiceError(AppError): def init(self, service: str, detail: str): super().init(f"{service} error: {detail}", code="EXTERNAL_ERROR", status=502)

### Error Classification

| Category | Retry? | Log Level | Alert? |
|----------|--------|-----------|--------|
| Validation error | No | WARNING | No |
| Not found | No | INFO | No |
| Auth failure | No | WARNING | Rate-based |
| Transient external | Yes | WARNING | After retries |
| Persistent external | No | ERROR | Yes |
| Internal bug | No | CRITICAL | Immediate |

## Structured Logging

### Setup with structlog

```python
import structlog
import logging

def configure_logging(log_level: str = "INFO", json_output: bool = True):
    processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
    ]

    if json_output:
        processors.append(structlog.processors.JSONRenderer())
    else:
        processors.append(structlog.dev.ConsoleRenderer())

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger(
            getattr(logging, log_level.upper())
        ),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
    )

log = structlog.get_logger()

Contextual Logging

import structlog
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="")

# Bind context per request
structlog.contextvars.bind_contextvars(
    request_id=request_id,
    user_id=user.id,
    organ="IV",
)

# All subsequent log calls include this context
log.info("processing_request", path="/api/skills", method="GET")
# Output: {"event": "processing_request", "request_id": "abc-123", "user_id": "u42", "path": "/api/skills", ...}

Log Levels by Purpose

Level	Purpose	Example
DEBUG	Detailed flow tracing	Query parameters, cache hits
INFO	Business events	User created, skill activated
WARNING	Recoverable issues	Retry attempt, deprecated usage
ERROR	Failures needing attention	External service down, data inconsistency
CRITICAL	System-level failures	Database unreachable, out of memory

Request Middleware

FastAPI Correlation IDs

import uuid
from starlette.middleware.base import BaseHTTPMiddleware

class CorrelationMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        structlog.contextvars.bind_contextvars(request_id=request_id)

        log.info("request_started",
            method=request.method,
            path=request.url.path,
        )

        try:
            response = await call_next(request)
            log.info("request_completed",
                status=response.status_code,
            )
            response.headers["X-Request-ID"] = request_id
            return response
        except Exception as e:
            log.error("request_failed", error=str(e), exc_info=True)
            raise
        finally:
            structlog.contextvars.unbind_contextvars("request_id")

Error Response Handler

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    log.warning("app_error", code=exc.code, message=exc.message)
    return JSONResponse(
        status_code=exc.status,
        content={"error": {"code": exc.code, "message": exc.message}},
    )

@app.exception_handler(Exception)
async def unhandled_error_handler(request: Request, exc: Exception):
    log.error("unhandled_error", error=str(exc), exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"error": {"code": "INTERNAL_ERROR", "message": "An unexpected error occurred"}},
    )

Logging Patterns

Operation Logging

async def process_skill(skill_id: str):
    log.info("skill_processing_started", skill_id=skill_id)
    try:
        result = await validate_skill(skill_id)
        log.info("skill_processing_completed", skill_id=skill_id, status=result.status)
        return result
    except ValidationError as e:
        log.warning("skill_validation_failed", skill_id=skill_id, error=e.message)
        raise
    except Exception as e:
        log.error("skill_processing_failed", skill_id=skill_id, error=str(e), exc_info=True)
        raise

Sensitive Data Filtering

SENSITIVE_KEYS = {"password", "token", "secret", "api_key", "authorization"}

def sanitize_log_data(data: dict) -> dict:
    return {
        k: "***REDACTED***" if k.lower() in SENSITIVE_KEYS else v
        for k, v in data.items()
    }

Performance Logging

import time
from contextlib import contextmanager

@contextmanager
def log_duration(operation: str, **extra):
    start = time.perf_counter()
    try:
        yield
    finally:
        duration_ms = (time.perf_counter() - start) * 1000
        log.info(f"{operation}_duration", duration_ms=round(duration_ms, 2), **extra)

Log Aggregation Integration

JSON Format for Ingestion

{
  "timestamp": "2026-03-20T10:58:00Z",
  "level": "info",
  "event": "request_completed",
  "request_id": "abc-123",
  "method": "GET",
  "path": "/api/skills",
  "status": 200,
  "duration_ms": 42.5,
  "service": "a-i--skills",
  "organ": "IV"
}

Common Fields

Always include: timestamp, level, event, service, request_id. Optionally: user_id, organ, duration_ms, error.

Anti-Patterns

• Logging PII or secrets — Always filter sensitive fields before logging
• String interpolation in log calls — Use structured fields: log.info("x", user=id) not log.info(f"user {id}")
• Catching and silencing exceptions — Log or re-raise; never except: pass
• Inconsistent log levels — Define team conventions and stick to them
• No correlation IDs — Impossible to trace requests across services without them
• Logging in tight loops — Rate-limit or sample high-frequency events