Architecting Layered FastAPI Applications

Hello! In the previous lesson, you coordinated independent async lookups and chose which failures should stop the combined operation. Now consider where that coordination—and the rules for interpreting its result—belongs in a FastAPI application. If it all lives in a route handler, even a small API can become difficult to review and test.

This lesson separates an incident API into routers, application services, domain models, and infrastructure adapters. You’ll examine the boundaries, then assemble a small working example. Allow about 40 minutes, including the selected video segments, readings, and code walkthrough.

Give each kind of change a home

Suppose a client requests an incident. FastAPI must interpret the path parameter and produce an HTTP response. The application must find the incident or report that it is missing. The incident itself may have a rule for whether it needs attention. A storage component must know how to retrieve it. Those are related tasks, but they change for different reasons.

A useful division is:

• Router: HTTP paths, request and response schemas, and translation of application errors into HTTP responses.
• Application service: The steps of a use case, such as retrieving an incident and deciding what to do if it does not exist.
• Domain model: Incident concepts and rules that make sense without an API or database.
• Infrastructure adapter: A concrete way to perform external work, such as reading from memory now or PostgreSQL later.

The concentric illustration below places domain concepts near the center and technology-specific components farther out. Treat it as a map of responsibilities, not as a literal path that an HTTP request travels through every ring.

Watch the selected portions of ArjanCodes’ Stop Mixing FastAPI with Business Logic: Fix It with Ports & Adapters. Its order-processing example uses a different domain, but the misplaced HTTP and database concerns are the same ones to watch for in an incident API.

The direction of Python imports matters more than the names of the folders. An incident model should not import FastAPI or a database model. A router may know about application services and domain values because it must translate them for HTTP. An infrastructure adapter may know about domain values because it must return them. The service should not need to know which database implementation was chosen.

Routers organize HTTP; services organize use cases

FastAPI’s APIRouter lets a module own a group of endpoints while the main application registers that group. Read the relevant portions of the official Bigger Applications – Multiple Files guide. Its example primarily separates routes into files; our example will add a service, domain, and adapter behind that routing structure.

A router is not an application service merely because it lives in another file. If a route contains incident rules, database queries, and persistence decisions, moving it to routers/incidents.py has improved navigation but not separation.

Architecture Patterns with Python calls the service layer an orchestration layer: it obtains data, invokes domain behavior, and handles the outcome. Its example uses Flask rather than FastAPI, but that distinction is independent of the web framework.

For our example, “is this open, critical incident in need of attention?” is a domain rule. “Find incident 42, or report that it was not found” is an application use case. “Return status 404” is an HTTP decision. Keeping those statements separate lets each be inspected without reading unrelated framework code.

Assemble a small incident API

Use the installable src-layout project established earlier in the course. The following example calls its package incident_api; adapt that package name if yours differs. Keep the __init__.py files in each package directory. No database is required for this walkthrough.

src/incident_api/
├── __init__.py
├── main.py
├── api/
│   ├── __init__.py
│   └── incidents.py
├── application/
│   ├── __init__.py
│   └── incidents.py
├── domain/
│   ├── __init__.py
│   └── incident.py
└── infrastructure/
    ├── __init__.py
    └── memory_incidents.py

Start with a domain model. It uses familiar dataclasses and enums; it has no knowledge of URLs, Pydantic, or storage. The attention rule is intentionally small so that its placement, rather than its complexity, is clear.

domain/incident.py

from dataclasses import dataclass
from enum import Enum


class Severity(str, Enum):
    LOW = "low"
    CRITICAL = "critical"


class Status(str, Enum):
    OPEN = "open"
    CLOSED = "closed"


@dataclass(frozen=True)
class Incident:
    id: int
    severity: Severity
    status: Status

    def needs_attention(self) -> bool:
        return self.status is Status.OPEN and self.severity is Severity.CRITICAL

Next, the application service expresses the retrieval use case. It receives a function that can retrieve an incident; it does not choose or construct the storage implementation. IncidentNotFound is meaningful to the application even when there is no HTTP request—for example, if a CLI asks for an unknown incident.

application/incidents.py

from collections.abc import Awaitable, Callable

from incident_api.domain.incident import Incident

GetIncident = Callable[[int], Awaitable[Incident | None]]


class IncidentNotFound(Exception):
    pass


class IncidentService:
    def __init__(self, get_incident: GetIncident) -> None:
        self._get_incident = get_incident

    async def read_incident(self, incident_id: int) -> Incident:
        incident = await self._get_incident(incident_id)
        if incident is None:
            raise IncidentNotFound(f"Incident {incident_id} not found")
        return incident

GetIncident is a deliberately small boundary: given an ID, the service can await an incident or None. It says nothing about a dictionary, SQL query, or HTTP client. The next lesson will replace this single-function shape with a typed repository contract that can describe a fuller set of operations.

For now, an infrastructure adapter keeps incidents in memory. It returns domain Incident objects, not API response models.

infrastructure/memory_incidents.py

from incident_api.domain.incident import Incident


class MemoryIncidentRepository:
    def __init__(self, incidents: list[Incident]) -> None:
        self._incidents = {incident.id: incident for incident in incidents}

    async def get(self, incident_id: int) -> Incident | None:
        return self._incidents.get(incident_id)

The async method matches the service’s awaited retrieval interface, but this particular dictionary lookup performs no I/O and does not yield while looking up an item. Later, an async database adapter can await actual non-blocking database operations. Merely declaring a method async would not make a blocking database driver non-blocking.

The router handles the HTTP boundary. Its Pydantic model describes what clients receive; it is separate from the internal dataclass. The route translates a missing-incident error into HTTP 404 and maps domain values into the response.

api/incidents.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

from incident_api.application.incidents import (
    IncidentNotFound,
    IncidentService,
)


class IncidentResponse(BaseModel):
    id: int
    severity: str
    status: str
    needs_attention: bool


def create_incident_router(service: IncidentService) -> APIRouter:
    router = APIRouter(prefix="/incidents", tags=["incidents"])

    @router.get("/{incident_id}", response_model=IncidentResponse)
    async def read_incident(incident_id: int) -> IncidentResponse:
        try:
            incident = await service.read_incident(incident_id)
        except IncidentNotFound as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

        return IncidentResponse(
            id=incident.id,
            severity=incident.severity.value,
            status=incident.status.value,
            needs_attention=incident.needs_attention(),
        )

    return router

Notice what the router does not do: it neither searches a dictionary nor defines the attention rule. Conversely, the service never raises HTTPException. If “not found” needs a different representation in another interface, only that interface’s translation changes.

Finally, main.py is the composition point. It chooses a concrete adapter, gives its get method to the service, and registers the router. Here we wire the objects explicitly so the boundary is visible; a later lesson will use FastAPI dependencies for more flexible wiring.

main.py

from fastapi import FastAPI

from incident_api.api.incidents import create_incident_router
from incident_api.application.incidents import IncidentService
from incident_api.domain.incident import Incident, Severity, Status
from incident_api.infrastructure.memory_incidents import MemoryIncidentRepository

repository = MemoryIncidentRepository(
    [
        Incident(id=42, severity=Severity.CRITICAL, status=Status.OPEN),
        Incident(id=43, severity=Severity.LOW, status=Status.CLOSED),
    ]
)
service = IncidentService(repository.get)

app = FastAPI()
app.include_router(create_incident_router(service))

With FastAPI and an ASGI server installed in your project environment, run the application from the project root:

python -m uvicorn incident_api.main:app --reload

A request to /incidents/42 should return an open, critical incident with "needs_attention": true. /incidents/43 should return false; /incidents/99 should return 404. This in-memory data is process-local, not persistent storage: restarting the process restores the two initial incidents.

Read the boundaries, not just the folders

Trace one successful request through the example. The router accepts an integer ID and calls the service. The service asks its supplied retrieval function for the incident. The memory adapter obtains a domain object. On return, the router evaluates the domain rule and builds the public response. A missing object takes a different path: the service raises IncidentNotFound, and the router assigns its HTTP meaning.

Now consider the earlier concurrent-summary use case. Its scheduling and fallback choices could live in an application service, while individual external calls would live in adapters. That placement would let a service test supply controlled collaborators instead of making live network calls. It would not change the TaskGroup failure behavior you learned last time; it would make the policy easier to locate and verify.

This is a useful code-review test for an API you maintain: Could the incident rule be run without FastAPI, and could the use case be run without the real storage system? If not, look at imports and responsibilities before adding more folders. Conversely, do not split every three-line function into a new layer merely to satisfy a diagram. The boundary earns its place when it keeps HTTP, use-case decisions, domain rules, and technology-specific work from becoming entangled.

Takeaways

A router owns HTTP concerns; an application service coordinates a use case; a domain model expresses incident concepts and rules; an infrastructure adapter performs a concrete retrieval or other external operation. APIRouter makes endpoints modular, but separation of concerns depends on what the modules import and do, not just where the files sit. In the working example, main.py is the one place that chooses the concrete adapter and connects the parts.

The service currently accepts one retrieval callable. Next, you’ll define a typed repository contract with Protocol, so a growing set of storage operations can remain explicit without making the service depend on PostgreSQL.