FastAPI makes it easy to get an endpoint running in five minutes. Keeping that API fast, readable and maintainable at 50 endpoints and several developers is a different problem. This is the structure and the handful of habits I reach for when building a FastAPI service on PostgreSQL that has to handle real traffic.

Start with a structure that scales

A single main.py stops being fun quickly. Group code by feature, not by file type, so everything about orders lives together:

app/
  main.py            # app factory, middleware, router wiring
  config.py          # settings from environment variables
  db.py              # engine, session factory, dependency
  orders/
    models.py        # SQLAlchemy tables
    schemas.py       # Pydantic request/response models
    service.py       # business logic
    router.py        # HTTP layer only
  users/
    ...
tests/

The rule that pays off most: routers stay thin. A router parses the request, calls a service function, and returns the result. Business rules live in service.py, where they can be tested without HTTP.

One async engine, one session per request

With SQLAlchemy 2.x and asyncpg, create the engine once at import time and hand each request its own session through a dependency:

from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.config import settings

engine = create_async_engine(
    settings.database_url,          # postgresql+asyncpg://user:pass@host/db
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,             # drop dead connections instead of erroring
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session

Two details matter here. expire_on_commit=False stops SQLAlchemy from invalidating loaded objects after a commit, which would otherwise trigger lazy loads that fail in async code. And pool_size plus max_overflow should add up to less than PostgreSQL's max_connections divided by the number of app processes you run.

Keep models and schemas separate

The table definition and the API contract change for different reasons, so give them different classes:

from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, ConfigDict
from sqlalchemy import ForeignKey, Numeric, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"), index=True)
    total: Mapped[Decimal] = mapped_column(Numeric(12, 2))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())


class OrderOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    customer_id: int
    total: Decimal
    created_at: datetime

Returning OrderOut through response_model means you never leak a column by accident, and FastAPI documents the exact shape for free.

Paginate with a cursor, not an offset

OFFSET 100000 makes PostgreSQL walk and discard 100,000 rows before returning any. Keyset pagination uses an indexed column to jump straight to the next page:

from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter(prefix="/orders", tags=["orders"])


@router.get("", response_model=list[OrderOut])
async def list_orders(
    after_id: int | None = None,
    limit: int = Query(50, ge=1, le=200),
    session: AsyncSession = Depends(get_session),
):
    stmt = select(Order).order_by(Order.id).limit(limit)
    if after_id is not None:
        stmt = stmt.where(Order.id > after_id)
    return (await session.scalars(stmt)).all()

The client passes the last id it received as after_id. The cost stays flat no matter how deep into the table you are. Always cap limit so nobody can request a million rows.

Watch for N+1 queries

The classic async trap: you load 50 orders, then access order.customer in a loop, and SQLAlchemy runs 50 extra queries (or raises, because lazy loading is not allowed in async). Tell it up front what to load:

from sqlalchemy.orm import selectinload

stmt = select(Order).options(selectinload(Order.customer)).limit(50)

selectinload issues one extra query for all customers instead of one per order. When something feels slow, turn on echo=True on the engine in development and count the statements per request.

Index for the questions you ask

Every filter and sort in a hot endpoint should be backed by an index. Check with EXPLAIN ANALYZE rather than guessing:

EXPLAIN ANALYZE
SELECT * FROM orders WHERE customer_id = 42 ORDER BY created_at DESC LIMIT 20;

If you see a sequential scan on a large table, a composite index such as (customer_id, created_at DESC) is usually the fix.

Do not block the event loop

An async def endpoint that calls a blocking library, such as a synchronous HTTP client or a CPU-heavy loop, freezes every other request on that worker. Use async clients (httpx.AsyncClient), or declare the endpoint with plain def so FastAPI runs it in a thread pool, and push genuinely heavy work to a background worker queue.

Run it properly

In production, run several worker processes behind a reverse proxy:

gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4 --bind 0.0.0.0:8000

A rough starting point is one worker per CPU core. Add a /health endpoint that runs SELECT 1 so your load balancer can tell when the database connection is unhealthy, and manage schema changes with Alembic migrations rather than create_all.

Test against a real database

Mocks hide the bugs that matter here: query shapes, constraints and transactions. Run tests against a throwaway PostgreSQL (a container works well), override the get_session dependency in the test client, and wrap each test in a transaction you roll back. They stay fast and they exercise the same code paths as production.

The short version

  • Group code by feature and keep routers thin.
  • One engine, one session per request, expire_on_commit=False.
  • Separate table models from API schemas.
  • Use keyset pagination and cap page sizes.
  • Load relations explicitly and count your queries.
  • Index for real queries, and verify with EXPLAIN ANALYZE.
  • Never block the event loop.

None of this is exotic. It is just the set of boring decisions that keep an API predictable as it grows.