Structure, dependencies, and schemas

Layout

Feature-first once the service is real:

app/
  main.py                   app factory, middleware, router registration only
  core/
    config.py               Settings (pydantic-settings)
    security.py             hashing, token encode/decode
    errors.py               domain exception types + handlers
    logging.py              structured logging setup
  db/
    session.py              engine/client, session dependency
    migrations/             alembic, or versioned scripts
  auctions/
    router.py               HTTP surface
    service.py              business rules
    repository.py           queries
    schemas.py              request/response models
    models.py               persistence models
  tests/

main.py should be boring: build the app, attach middleware, include routers, define lifespan. If it grows business logic or long import lists of helpers, the features are leaking upward.

Settings

One Settings object, validated at startup, injected rather than imported:

from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    database_url: str
    jwt_secret: str
    jwt_ttl_seconds: int = 900
    cors_origins: list[str] = []
    environment: str = "development"

@lru_cache
def get_settings() -> Settings:
    return Settings()

Why this shape:

• Fails fast. A missing or malformed variable stops the process at boot

instead of raising a KeyError at 3am on one code path.

• lru_cache makes it a singleton without a global.
• Injectable, so tests override it via dependency_overrides instead of

monkeypatching os.environ.

• Never read os.getenv deep in a service. That is an untestable hidden input.

Dependencies

Dependencies are the composition root. Use them for anything cross-cutting:

async def get_db() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session                      # teardown runs after the response

async def current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User: ...

def require_role(*roles: str):
    async def guard(user: Annotated[User, Depends(current_user)]) -> User:
        if user.role not in roles:
            raise Forbidden("role not permitted")
        return user
    return guard

@router.delete("/{id}", dependencies=[Depends(require_role("admin"))])
async def delete_auction(...): ...

A yield dependency guarantees cleanup even when the handler raises. This is how sessions and connections should always be managed; a try/finally inside each handler is the version that eventually gets forgotten.

require_role as a factory keeps authorization declarative and greppable. Permission checks buried in handler bodies are how endpoints quietly ship without them.

Schemas: three models, not one

class VehicleCreate(BaseModel):     # what a client may send
    make: str
    model: str
    price_minor: int = Field(gt=0)

class VehicleUpdate(BaseModel):     # everything optional
    make: str | None = None
    price_minor: int | None = Field(default=None, gt=0)

class VehicleOut(BaseModel):        # what you promise to return
    id: str
    make: str
    price_minor: int
    model_config = ConfigDict(from_attributes=True)

Reusing one model for all three is the most common source of two bugs: fields required on update that should not be, and internal fields leaking to clients because the model that describes the row is also the model that serializes the response.

Never return a database object directly. Declare response_model and let FastAPI filter. That is what stops a password_hash column reaching an API consumer the day someone adds it.

Money is an integer in minor units. Floats for currency produce rounding disputes that are painful to unwind after the fact.

Errors

Raise domain exceptions in services; translate at the edge.

# core/errors.py
class DomainError(Exception): ...
class NotFound(DomainError): ...
class Conflict(DomainError): ...
class Forbidden(DomainError): ...

STATUS = {NotFound: 404, Conflict: 409, Forbidden: 403}

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(
        status_code=STATUS.get(type(exc), 400),
        content={"error": {"code": type(exc).__name__, "message": str(exc)}},
    )

The service layer stays free of HTTPException, which is what lets the same code run from a worker or a CLI. One envelope shape across every error means clients write one handler.

Add a RequestValidationError handler so 422 responses match that envelope too; FastAPI's default shape differs from whatever you designed, and clients notice.

Never put exception text from the database or an upstream provider into a response body. Log it with the request id and return something generic.

Pagination

Offset pagination degrades: OFFSET 100000 makes the database walk and discard 100,000 rows, and a row inserted mid-scroll shifts every page.

Prefer cursor (keyset) pagination on anything unbounded:

GET /v1/auctions?limit=50&cursor=eyJpZCI6ICI2NWYuLi4ifQ

Order by an indexed, unique, monotonic column and return the last key as an opaque cursor. Keep offset only for small admin tables where a page-number UI is genuinely wanted.

Always cap limit. An uncapped one is a denial-of-service endpoint.

N+1

The most common FastAPI performance bug, and it never shows up on a seeded dev database with twelve rows.

• SQL: eager-load with selectinload, or fetch the related set in one query and

join in memory.

• Document stores: $lookup, or a second batched query keyed by id.
• Assert it. A test that counts queries around a list endpoint catches the

regression the day it lands rather than the day traffic arrives.

Background work

BackgroundTasks is fine for fire-and-forget work that may be lost: a best-effort email, a cache warm.

It is not a queue. It runs in the same process, dies with it, has no retries and no visibility. Anything that must happen belongs in a real queue with a durable store, or in a scheduled job.

Scheduled sweeps (expiring auctions, retrying payments) should be callable as plain service functions so a test can invoke them directly rather than waiting on a timer.