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

Structured Logging

Setup with structlog

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

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