Custom Exception Handling

Map domain exceptions to appropriate HTTP responses.

Goal

Convert service-level exceptions into meaningful HTTP error responses.

Prerequisites

• A controller extending BaseAsyncController or BaseCeleryTaskController
• Domain exceptions defined in exceptions.py

The Pattern

Override handle_exception() in your controller:

from typing import Any

from fastapi import HTTPException, status


async def handle_exception(self, exception: Exception) -> Any:
    if isinstance(exception, YourDomainError):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(exception),
        ) from exception
    return await super().handle_exception(exception)

Step-by-Step

1. Define Domain Exceptions

Create exceptions in the domain exception module:

# src/fastdjango/core/order/exceptions.py
from fastdjango.core.exceptions import ApplicationError


class OrderNotFoundError(ApplicationError):
    """Raised when an order cannot be found."""


class OrderAlreadyPaidError(ApplicationError):
    """Raised when trying to pay an already paid order."""


class InsufficientStockError(ApplicationError):
    """Raised when stock is insufficient for order."""


class InvalidOrderStateError(ApplicationError):
    """Raised when order operation is invalid for current state."""

2. Raise Exceptions in Service

from asgiref.sync import sync_to_async
from diwire import Injected

from fastdjango.foundation.services import BaseService
from fastdjango.foundation.transactions import TransactionFactory
from fastdjango.core.order.exceptions import (
    InsufficientStockError,
    InvalidOrderStateError,
    OrderAlreadyPaidError,
    OrderNotFoundError,
)

@dataclass(kw_only=True)
class OrderService(BaseService):
    _transaction_factory: Injected[TransactionFactory]

    async def get_order_by_id(self, *, order_id: int) -> Order:
        try:
            return await Order.objects.aget(id=order_id)
        except Order.DoesNotExist as e:
            raise OrderNotFoundError(f"Order {order_id} not found") from e

    async def pay_order(self, *, order_id: int) -> Order:
        return await sync_to_async(
            self._pay_order_transactionally,
            thread_sensitive=True,
        )(order_id=order_id)

    def _pay_order_transactionally(self, *, order_id: int) -> Order:
        with self._transaction_factory(span_name="pay order"):
            try:
                order = Order.objects.get(id=order_id)
            except Order.DoesNotExist as e:
                raise OrderNotFoundError(f"Order {order_id} not found") from e

            if order.status == OrderStatus.PAID:
                raise OrderAlreadyPaidError(f"Order {order_id} is already paid")

            if order.status != OrderStatus.PENDING:
                raise InvalidOrderStateError(
                    f"Cannot pay order in {order.status} state"
                )

            order.status = OrderStatus.PAID
            order.save()
            return order

3. Map Exceptions in Controller

# src/fastdjango/core/order/delivery/fastapi/controllers.py
from dataclasses import dataclass
from typing import Any

from fastapi import APIRouter, HTTPException, status

from fastdjango.core.order.exceptions import (
    InsufficientStockError,
    InvalidOrderStateError,
    OrderAlreadyPaidError,
    OrderNotFoundError,
)
from fastdjango.core.order.services import (
    OrderService,
)
from fastdjango.foundation.delivery.controllers import BaseAsyncController


@dataclass(kw_only=True)
class OrderController(BaseAsyncController):
    _order_service: OrderService

    def register(self, registry: APIRouter) -> None:
        registry.add_api_route(
            path="/v1/orders/{order_id}/pay",
            endpoint=self.pay_order,
            methods=["POST"],
        )

    async def pay_order(self, order_id: int) -> OrderSchema:
        order = await self._order_service.pay_order(order_id=order_id)
        return OrderSchema.model_validate(order, from_attributes=True)

    async def handle_exception(self, exception: Exception) -> Any:
        # 404 - Resource not found
        if isinstance(exception, OrderNotFoundError):
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=str(exception),
            ) from exception

        # 409 - Conflict (already in desired state)
        if isinstance(exception, OrderAlreadyPaidError):
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=str(exception),
            ) from exception

        # 422 - Unprocessable (invalid state for operation)
        if isinstance(exception, InvalidOrderStateError):
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=str(exception),
            ) from exception

        # 400 - Bad request (insufficient stock)
        if isinstance(exception, InsufficientStockError):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=str(exception),
            ) from exception

        # Re-raise unknown exceptions
        return await super().handle_exception(exception)

Exception to HTTP Status Mapping

Exception Type	HTTP Status	When to Use
NotFoundError	404	Resource doesn't exist
AccessDeniedError	403	User can't access resource
AlreadyExistsError	409	Resource already exists
InvalidStateError	422	Operation invalid for current state
ValidationError	400	Input validation failed
InsufficientError	400	Not enough resources
UnauthorizedError	401	Authentication required/failed

Structured Error Responses

For more detailed error responses, create an error schema:

# src/fastdjango/core/common/delivery/fastapi/schemas.py
from fastdjango.foundation.delivery.fastapi.schemas import BaseFastAPISchema


class ErrorResponseSchema(BaseFastAPISchema):
    error: str
    code: str
    details: dict | None = None

Then use it in exception handling:

async def handle_exception(self, exception: Exception) -> Any:
    if isinstance(exception, InsufficientStockError):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail={
                "error": str(exception),
                "code": "INSUFFICIENT_STOCK",
                "details": {
                    "requested": exception.requested,
                    "available": exception.available,
                },
            },
        ) from exception

Multiple Exception Types

Handle multiple similar exceptions together:

async def handle_exception(self, exception: Exception) -> Any:
    # Group 404 errors
    not_found_errors = (
        OrderNotFoundError,
        ProductNotFoundError,
        UserNotFoundError,
    )
    if isinstance(exception, not_found_errors):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exception),
        ) from exception

    # Group permission errors
    permission_errors = (
        AccessDeniedError,
        InsufficientPermissionsError,
    )
    if isinstance(exception, permission_errors):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=str(exception),
        ) from exception

    return await super().handle_exception(exception)

Testing Exception Handling

@pytest.mark.django_db(transaction=True)
class TestOrderController:
    def test_pay_order_not_found(
        self,
        test_client_factory: TestClientFactory,
        user: User,
    ) -> None:
        with test_client_factory(auth_for_user=user) as client:
            response = client.post("/v1/orders/99999/pay")

        assert response.status_code == HTTPStatus.NOT_FOUND
        assert "not found" in response.json()["detail"].lower()

    def test_pay_order_already_paid(
        self,
        test_client_factory: TestClientFactory,
        user: User,
        paid_order: Order,
    ) -> None:
        with test_client_factory(auth_for_user=user) as client:
            response = client.post(f"/v1/orders/{paid_order.id}/pay")

        assert response.status_code == HTTPStatus.CONFLICT

Best Practices

1. Be specific: Use descriptive exception classes, not generic Exception
2. Include context: Pass relevant IDs and values in exception messages
3. Use appropriate status codes: Follow HTTP semantics
4. Chain exceptions: Use from exception to preserve stack trace
5. Call super(): Always call super().handle_exception() for unknown exceptions