Exception Hierarchy

Domain, application, infrastructure, and validation exception types — how to structure and organise them.

Show:

Status: Extracted
Purpose: Exception hierarchy design, exception types, and exception organization.

Exception Hierarchy

Exception Types

Domain Exceptions:

Business rule violations
Invariant violations
Domain-specific errors

Application Exceptions:

Use case failures
Orchestration errors
Application-level errors

Infrastructure Exceptions:

Database errors
External service errors
Infrastructure-level errors

Validation Exceptions:

Pydantic validation errors
Request validation errors
Input validation errors

Exception Organization

Exception Location

API Exceptions:

Location: api/exceptions.py
HTTP-related exceptions
API-specific errors

Domain Exceptions:

Location: domain/{aggregate}/exceptions.py
Domain-specific errors
Business rule violations

Application Exceptions:

Location: application/exceptions.py or application/{domain}/exceptions.py
Application-level errors
Use case failures

Infrastructure Exceptions:

Location: infrastructure/{component}/exceptions.py
Infrastructure-specific errors
External service errors

Exception Examples

Domain Exceptions

Pattern:

# domain/account/exceptions.py
class AccountNotFoundError(Exception):
    """Account not found error."""
    
    def __init__(self, account_id: str):
        self.account_id = account_id
        super().__init__(f"Account with ID '{account_id}' not found")
 
class AccountNotAuthorizedError(Exception):
    """Account not authorized error."""
    
    def __init__(self, account_id: str, reason: str):
        self.account_id = account_id
        self.reason = reason
        super().__init__(f"Account '{account_id}' not authorized: {reason}")

Application Exceptions

Pattern:

# application/exceptions.py
class UseCaseError(Exception):
    """Base exception for use case errors."""
    pass
 
class OrchestrationError(UseCaseError):
    """Orchestration error."""
    pass

Infrastructure Exceptions

Pattern:

# infrastructure/database/exceptions.py
class DatabaseError(Exception):
    """Base exception for database errors."""
    pass
 
class ConnectionError(DatabaseError):
    """Database connection error."""
    pass

Validation Exceptions

Pattern:

# api/exceptions.py
class ValidationError(Exception):
    """Validation error."""
    
    def __init__(self, message: str, field: str | None = None):
        self.field = field
        super().__init__(message)

Exception Best Practices

Exception Design

Specific exceptions: Create specific exceptions for different error cases
Exception messages: Clear, user-friendly error messages
Exception context: Include relevant context (IDs, fields, etc.)
Exception hierarchy: Use inheritance for related exceptions

Exception Usage

Throw exceptions: Use exceptions for exceptional cases
Don't use for control flow: Exceptions are for errors, not normal flow
Handle at boundaries: Handle exceptions at layer boundaries (API layer)
Log exceptions: Log all exceptions with context

Error Hierarchy

Exception
└── DomainError                        # base for all domain errors (domain/)
    ├── EntityNotFoundError            # entity does not exist
    ├── NotAuthorizedError             # actor lacks permission
    ├── ValidationError                # business rule violated
    └── ConflictError                  # e.g. duplicate, state conflict

└── ApplicationError                   # base for application-layer errors (application/)
    └── (add as needed)

└── InfrastructureError                # base for infra errors (infrastructure/)
    └── (add as needed)

Where each type is raised:

DomainError subclasses — raised in domain/ and application/ handlers. Never raised in api/ or infrastructure/.
ValidationError (from Pydantic) — raised by FastAPI automatically on request schema failures.
InfrastructureError — raised in infrastructure/ adapters. Should be caught and wrapped in ApplicationError at the application boundary if they need to propagate.

Domain Error Base (`domain/errors.py`)

class DomainError(Exception):
    """Base for all domain errors. Carries a machine-readable error code."""
    def __init__(self, message: str, error_code: str, is_retryable: bool = False):
        super().__init__(message)
        self.error_code = error_code
        self.is_retryable = is_retryable
 
 
class EntityNotFoundError(DomainError):
    def __init__(self, entity_type: str, entity_id: str):
        super().__init__(
            message=f"{entity_type} not found: {entity_id}",
            error_code=f"{entity_type.upper()}_NOT_FOUND",
        )
        self.entity_type = entity_type
        self.entity_id = entity_id
 
 
class NotAuthorizedError(DomainError):
    def __init__(self, reason: str):
        super().__init__(message=reason, error_code="NOT_AUTHORIZED")
 
 
class ConflictError(DomainError):
    def __init__(self, message: str):
        super().__init__(message=message, error_code="CONFLICT")

Exception Handlers (`api/middleware/error_handler.py`)

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from domain.errors import EntityNotFoundError, NotAuthorizedError, ConflictError, DomainError
import logging, traceback
 
logger = logging.getLogger(__name__)
 
 
def register_exception_handlers(app: FastAPI) -> None:
 
    @app.exception_handler(EntityNotFoundError)
    async def not_found_handler(request: Request, exc: EntityNotFoundError):
        _log(logger.warning, "entity.not_found", exc, request)
        return JSONResponse(
            {"detail": str(exc), "error_code": exc.error_code},
            status_code=404,
        )
 
    @app.exception_handler(NotAuthorizedError)
    async def not_authorized_handler(request: Request, exc: NotAuthorizedError):
        _log(logger.warning, "not.authorized", exc, request)
        return JSONResponse(
            {"detail": str(exc), "error_code": exc.error_code},
            status_code=403,
        )
 
    @app.exception_handler(ConflictError)
    async def conflict_handler(request: Request, exc: ConflictError):
        _log(logger.warning, "conflict", exc, request)
        return JSONResponse(
            {"detail": str(exc), "error_code": exc.error_code},
            status_code=409,
        )
 
    @app.exception_handler(RequestValidationError)
    async def validation_handler(request: Request, exc: RequestValidationError):
        return JSONResponse(
            {"detail": exc.errors(), "error_code": "VALIDATION_ERROR"},
            status_code=422,
        )
 
    @app.exception_handler(Exception)
    async def catchall_handler(request: Request, exc: Exception):
        ctx = getattr(request.state, "ctx", {})
        logger.error(
            "unhandled.exception",
            extra={
                "request_id": ctx.get("request_id"),
                "exception_type": type(exc).__name__,
                "exception_message": str(exc),
                "traceback": traceback.format_exc(),
                "error_code": "INTERNAL_SERVER_ERROR",
                "is_retryable": False,
            },
            exc_info=True,
        )
        return JSONResponse(
            {
                "detail": "Internal server error",
                "request_id": ctx.get("request_id"),
                "error_code": "INTERNAL_SERVER_ERROR",
            },
            status_code=500,
        )
 
 
def _log(log_fn, event: str, exc: Exception, request: Request) -> None:
    ctx = getattr(request.state, "ctx", {})
    log_fn(event, extra={
        "request_id": ctx.get("request_id"),
        "error_code": getattr(exc, "error_code", None),
        "exception_message": str(exc),
    })

Error Response Shape

All error responses follow the same JSON shape:

{
  "detail": "Human-readable message",
  "error_code": "MACHINE_READABLE_CODE",
  "request_id": "550e8400-..."
}

request_id is included on 5xx errors (for support tracing). error_code is included on all errors.

Webhook Special Case

Webhooks should always return 200 to prevent the sending provider from retrying. Add path-based routing in the catch-all handler:

@app.exception_handler(Exception)
async def catchall_handler(request: Request, exc: Exception):
    if request.url.path.startswith("/webhooks/"):
        logger.warning("webhook.error", extra={"path": request.url.path, "error": str(exc)})
        return JSONResponse({"status": "received"}, status_code=200)
    # ... normal 500 handling

Rules

Domain and application layers raise typed exceptions — they never return error dicts or None to signal failure.
Route handlers do not catch exceptions — they let exception handlers do it.
Never raise HTTPException in application or domain layers. Raise domain errors; let the handler map them to HTTP codes.
Every domain error has a machine-readable error_code string. This is what client applications key off — not the HTTP status code or the human message.
is_retryable on domain errors tells consumers whether it's worth retrying the operation.

CQRS — Commands & Queries Unit of Work Pattern

Exception Hierarchy