Python + FastAPI + SQLAlchemy

Project context

This is a FastAPI service with async SQLAlchemy 2.0 and Postgres. Migrations via Alembic. Pydantic v2 for I/O validation. Tests with pytest-asyncio. Production target.

Stack

• Python 3.12+
• FastAPI 0.110+
• SQLAlchemy 2.0+ (async, declarative)
• Alembic for migrations
• Postgres 15+ (asyncpg driver)
• Pydantic v2 for request/response models
• uv for package management (preferred over pip / poetry)
• pytest + pytest-asyncio for tests
• Ruff for lint + format
• mypy in strict mode

Folder structure

app/
  main.py              — FastAPI app factory
  config.py            — Pydantic Settings
  api/
    __init__.py
    routes/            — route modules, one per resource
    deps.py            — dependency-injection helpers
  db/
    base.py            — SQLAlchemy declarative base
    session.py         — engine + session factory
    models/            — ORM models, one file per table
  schemas/             — Pydantic request/response models
  services/            — business logic (orchestration of repo + external APIs)
  repos/               — data access (SQLAlchemy queries)
alembic/
  versions/
  env.py
tests/
  unit/
  integration/
  conftest.py

Async-first

Everything is async — routes, services, repos, DB calls. Don't mix sync and async DB sessions; the engine is create_async_engine and sessions are AsyncSession.

# app/db/session.py
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine(settings.database_url, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

Pydantic v2 models

• Separate request and response schemas — don't reuse the ORM model
• Use model_config = ConfigDict(from_attributes=True) on response models that map from ORM
• Use Field(...) for constraints (min_length, max_length, gt, le)
• Use Annotated for complex types

class PostCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    body: str

class PostRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    body: str
    created_at: datetime

Routes

@router.post("/posts", response_model=PostRead, status_code=201)
async def create_post(
    payload: PostCreate,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> Post:
    return await post_service.create(db, payload, user)

• response_model on every route — never let FastAPI auto-derive from return type alone
• status_code explicit on every non-200
• Use Depends() for DB session, current user, settings

Dependency injection

• app/api/deps.py holds reusable dependencies: get_db, get_current_user, require_admin
• Override dependencies in tests via app.dependency_overrides[get_db] = ...
• Don't reach for global state — pass via Depends

Database queries (SQLAlchemy 2.0 style)

from sqlalchemy import select

async def list_posts(db: AsyncSession, *, limit: int = 20) -> list[Post]:
    stmt = select(Post).order_by(Post.created_at.desc()).limit(limit)
    result = await db.execute(stmt)
    return list(result.scalars().all())

• Always use select() — not the legacy Query API
• Use .scalars() to unwrap; .scalar_one() if you expect exactly one
• selectinload, joinedload for eager-loading relationships — avoid N+1
• Wrap multi-step writes in async with db.begin() for explicit transactions

Migrations

• alembic revision --autogenerate -m "<name>" — generate
• Review the SQL — autogenerate misses some changes (especially constraints)
• alembic upgrade head — apply locally
• alembic downgrade -1 — roll back one
• Never edit a committed migration; create a new one

Errors

• Raise HTTPException(status_code=...) for HTTP-shaped errors
• For domain errors, raise typed exceptions in services and translate at the route boundary
• Don't expose internal error messages to clients — use a structured error response

Logging

• Structured logging via structlog or stdlib + JSON formatter
• One log per request at INFO; DEBUG for noisy internals
• Never log secrets, tokens, or PII without masking

Testing

• pytest-asyncio mode = auto
• Use httpx.AsyncClient against the FastAPI app — no real server
• Override get_db to use a transaction that rolls back per test
• Factory-style fixtures with pytest-factoryboy for ORM models
• Aim for integration tests at the route boundary; unit tests for service logic

@pytest.fixture
async def client(db):
    app.dependency_overrides[get_db] = lambda: db
    async with AsyncClient(app=app, base_url="http://test") as c:
        yield c

Patterns to avoid

• Sync DB calls in async routes — they block the event loop
• session.query(...) — legacy API; use select(...)
• Sharing one session across requests — always per-request via Depends
• Returning ORM models directly — always convert to Pydantic response models (or use response_model)
• Catching broad Exception at route level — let FastAPI's exception handlers take it
• Hand-rolled JWT logic — use python-jose or authlib

Tooling

• uv pip install -r requirements.txt
• uvicorn app.main:app --reload — dev server
• pytest — tests
• ruff check && ruff format — lint + format
• mypy app/ — type check
• alembic upgrade head — apply migrations

AI behavioral rules

• Default to async — never mix sync and async DB code
• Use select(...) for queries; never the legacy query() API
• Pydantic response models on every route, separate from the ORM model
• Never modify an Alembic migration after it's been committed
• Don't catch Exception broadly; let FastAPI's handlers translate
• Validate inputs with Pydantic at the route boundary; don't trust upstream input
• Run pytest, ruff check, and mypy before declaring a task done