SearchBook a call

RESEARCH / ARCHIVE

FastAPI, done properly

The layout that survives year two, the async rule that takes services down, where Pydantic belongs, and the five ways it bites in production.

From the archive. This page preserves the original publication; product plans and capabilities may have changed. See current open source work.

FastAPI is the easiest Python web framework to start badly. The tutorial gets you a working endpoint in nine lines, and nothing warns you that the same shape, repeated four hundred times, becomes a service nobody wants to touch.

Know what you are using

FastAPI is thinner than it looks. It is Starlette for the ASGI application, routing, middleware, WebSockets and the test client, plus Pydantic for validation and serialisation, plus a dependency injection system and an OpenAPI generator.

This matters because when something breaks, the answer is usually in Starlette's docs or Pydantic's, not FastAPI's. Knowing which of the three layers you are fighting cuts debugging time.

A layout that survives the second year

The tutorial keeps everything in main.py. The common next step is to split by technical kind: a routers/ folder, a models/ folder, a schemas/ folder. That works until you have thirty features.

Split by feature, then by kind inside the feature:

app/
  main.py               app factory, lifespan, middleware, router mounting
  config.py             settings, one Pydantic BaseSettings, fail loud
  db.py                 engine, session factory, base
  api/deps.py           shared dependencies (session, current user, pagination)
  features/
    orders/
      router.py         HTTP only: parse, call service, shape response
      schemas.py        request and response models
      models.py         SQLAlchemy tables
      service.py        the actual logic, no FastAPI imports
      repository.py     queries, no business rules
    billing/
      ...
  workers/              Celery, arq, or whatever runs outside the request
tests/

The rule: service.py must not import anything from FastAPI. No Depends, no HTTPException, no Request.

Services raise domain errors; the router translates those into HTTP.

The async rule that matters

FastAPI runs on an event loop. A route declared async def runs on that loop. If that coroutine performs a blocking call, a synchronous database query, requests.get, time.sleep, a CPU-heavy loop, it blocks the entire loop, not just that request.

A route declared plain def is different: FastAPI runs it in a threadpool, so blocking there is contained. Which produces the counterintuitive rule:

If your handler does blocking work, def is safer than async def. The wrong async is far more dangerous than no async.

The failure mode does not look like a bug. Latency rises under load, p99 goes vertical, and CPU sits low because everything waits on one blocked thread.

Three ways out, in order of preference:

  1. Go async all the way down. asyncpg through SQLAlchemy's async engine, httpx.AsyncClient instead of requests.
  2. Stay synchronous and use plain def handlers. Boring, correct, and perfectly fast for the large majority of services.
  3. Mix carefully, pushing blocking calls through run_in_threadpool (or asyncio.to_thread) when you need an async handler that touches a sync library.

Do not write async def everywhere because it looks modern, then call a synchronous ORM inside it.

Dependency injection, used as a seam

Depends is FastAPI's best idea and its most abused feature. It is a seam: a router declares what it needs without knowing how to build it, and a test swaps the construction.

# api/deps.py
async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session

async def current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    session: Annotated[AsyncSession, Depends(get_session)],
) -> User:
    user = await users.by_token(session, token)
    if user is None:
        raise HTTPException(401, "Invalid credentials")
    return user

# features/orders/router.py
@router.post("/orders", response_model=OrderOut, status_code=201)
async def create_order(
    payload: OrderIn,
    user: Annotated[User, Depends(current_user)],
    session: Annotated[AsyncSession, Depends(get_session)],
):
    return await orders.create(session, user, payload)

First, use Annotated rather than default values; it is the current idiom and keeps the signature readable. Second, dependency overrides are the reason this is worth it:

app.dependency_overrides[get_session] = lambda: test_session
app.dependency_overrides[current_user] = lambda: fake_admin

You get integration tests without patching module internals.

Where it goes wrong is depth. A dependency that pulls three dependencies that pull four more becomes an invisible call graph that runs before your handler. Keep the chain shallow.

Pydantic at the boundary, not through the core

Pydantic v2 moved validation into Rust and is fast.

Validation belongs at the edges: the HTTP request in, the HTTP response out, configuration at boot.

Keep request and response models separate, always. Reuse one Order model for both and you must hide the internal cost field on the way out, or accept an id on the way in that you must ignore, and one class carries two contradictory contracts.

Configuration deserves the same rigour:

class Settings(BaseSettings):
    database_url: PostgresDsn
    redis_url: RedisDsn
    secret_key: SecretStr
    environment: Literal["local", "staging", "production"] = "local"

    model_config = SettingsConfigDict(env_file=".env", extra="forbid")

settings = Settings()   # raises at import if anything is missing or malformed

extra="forbid" catches the typo in an environment variable name that would otherwise silently fall back to a default.

Lifespan, and the connection pool you forgot

@app.on_event("startup") is deprecated. Use the lifespan context manager, which writes teardown next to setup:

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=10.0)
    app.state.redis = await create_redis_pool(settings.redis_url)
    yield
    await app.state.http.aclose()
    await app.state.redis.close()

app = FastAPI(lifespan=lifespan)

Create clients once, here. An httpx.AsyncClient constructed inside a handler opens a fresh connection pool per request, meaning a new TLS handshake per request.

The same applies to database pool sizing. The default pool is small. Under a process manager running several workers, your effective connection count is workers multiplied by pool size, and Postgres has a hard ceiling. Size it deliberately.

Background work: the trap in the standard library of the framework

BackgroundTasks looks like a job queue and is not one. It runs after the response is sent, in the same process. If the process restarts, deploys or crashes, the work is gone with no record and no retry.

It is correct for fire-and-forget work where loss is acceptable: emitting a metric, writing a non-critical audit line, warming a cache. It is wrong for sending the receipt email, charging the card, or syncing inventory.

The moment work must survive a deploy, it belongs in a real queue with durable storage and retries. Celery if you already have it, arq if you want async-native and small, or a Postgres-backed queue if you would rather not run another broker. A table with SELECT ... FOR UPDATE SKIP LOCKED is a respectable job queue and one less service to operate.

Testing

FastAPI's testability is its quietest advantage.

@pytest.fixture
async def client(session):
    app.dependency_overrides[get_session] = lambda: session
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

Because ASGITransport talks to the app in-process, there is no socket, no port, no server to start. Full request-to-database tests run in milliseconds, so you can afford to write them for real paths.

Test against a real Postgres, not SQLite. The dialects diverge on exactly the things that break in production: JSON operators, upserts, constraint behaviour, transaction semantics.

Where it bites in production

  • No timeouts on outbound calls. An httpx client without an explicit timeout will wait a very long time. One slow upstream then consumes every worker. Set a timeout on every client, at construction, in lifespan.
  • Unbounded response models. An endpoint that returns a list with no pagination is a table scan and a multi-megabyte payload. Paginate from the first commit; adding it later is an API break.
  • Exception handlers that leak. The default handler returns a 500 with a generic body. Never put an ORM error string in a response.
  • OpenAPI drift. The generated schema is only as good as your response_model declarations. An endpoint returning a bare dict documents itself as "an object".
  • Middleware order. Starlette applies middleware in reverse of registration. Get this wrong and your request-ID middleware runs after your logging middleware.

When not to reach for it

FastAPI is the right default for an API: a service that speaks JSON to a frontend, a mobile app, or another service.

It is the wrong default when you want a batteries-included application: server rendered pages, an admin interface, an ORM with migrations, auth, sessions and permissions out of the box. That is Django.