Python + FastAPI

1# Python + FastAPI
2 
3HTTP API in Python with FastAPI. Prioritize type safety via Pydantic v2, thin route handlers, service-layer business logic, and async-first patterns. Every change must pass `ruff check . && ruff format . --check && pytest` before committing.
4 
5## Quick Reference
6 
7| Task | Command |
8|---|---|
9| Run dev server | `uvicorn app.main:app --reload` |
10| Run all tests | `pytest` |
11| Run single test | `pytest tests/test_orders.py::test_create_order -x` |
12| Run tests (parallel) | `pytest -n auto` |
13| Lint | `ruff check .` |
14| Lint (fix) | `ruff check . --fix` |
15| Format | `ruff format .` |
16| Format (check only) | `ruff format . --check` |
17| Type check | `mypy .` |
18| Alembic: create migration | `alembic revision --autogenerate -m "description"` |
19| Alembic: migrate up | `alembic upgrade head` |
20| Alembic: migrate down one | `alembic downgrade -1` |
21| Alembic: show history | `alembic history --verbose` |
22| Docker up (full stack) | `docker compose up -d` |
23| Docker rebuild | `docker compose up -d --build` |
24| Install deps | `uv sync` or `pip install -e ".[dev]"` |
25 
26## Project Structure
27 
28```
29├── app/
30│   ├── __init__.py
31│   ├── main.py              # FastAPI app factory, middleware, lifespan
32│   ├── config.py            # Pydantic Settings — all env var parsing
33│   ├── dependencies.py      # Shared FastAPI dependencies (get_db, get_current_user)
34│   ├── database.py          # SQLAlchemy engine, sessionmaker, Base
35│   ├── models/              # SQLAlchemy ORM models
36│   │   ├── __init__.py
37│   │   ├── base.py          # Declarative base, common mixins (timestamps, UUID PK)
38│   │   ├── user.py
39│   │   └── order.py
40│   ├── schemas/             # Pydantic v2 request/response schemas
41│   │   ├── __init__.py
42│   │   ├── user.py
43│   │   └── order.py
44│   ├── services/            # Business logic — no HTTP or FastAPI imports
45│   │   ├── __init__.py
46│   │   ├── user.py
47│   │   └── order.py
48│   ├── routers/             # Route handlers — thin, delegate to services
49│   │   ├── __init__.py
50│   │   ├── user.py
51│   │   └── order.py
52│   ├── middleware/           # Custom middleware (logging, correlation ID)
53│   └── exceptions.py        # Domain exceptions + exception handlers
54├── alembic/
55│   ├── env.py
56│   └── versions/
57├── tests/
58│   ├── conftest.py          # Fixtures: test client, DB session, factories
59│   ├── factories.py         # Polyfactory or factory_boy factories
60│   ├── test_orders.py
61│   └── test_users.py
62├── alembic.ini
63├── pyproject.toml
64├── .env.example
65├── Dockerfile
66└── docker-compose.yml
67```
68 
69`app/routers/` → `app/services/` → `app/models/` is the dependency direction. Routers never import models directly; services are the bridge.
70 
71## Code Conventions
72 
73### Pydantic v2 Schemas
74 
75All request/response models use Pydantic v2. Never use Pydantic v1 patterns.
76 
77```python
78from pydantic import BaseModel, Field, ConfigDict
79 
80class OrderCreate(BaseModel):
81    model_config = ConfigDict(strict=True)
82 
83    items: list[OrderItemCreate] = Field(..., min_length=1)
84    shipping_address_id: uuid.UUID
85    note: str | None = Field(None, max_length=500)
86 
87class OrderResponse(BaseModel):
88    model_config = ConfigDict(from_attributes=True)
89 
90    id: uuid.UUID
91    status: OrderStatus
92    total_cents: int
93    created_at: datetime
94    items: list[OrderItemResponse]
95```
96 
97**Rules:**
98- Use `model_config = ConfigDict(from_attributes=True)` for ORM-backed responses — never the old `class Config: orm_mode = True`.
99- Use `Field(...)` for required fields with constraints. Use `Field(None)` for optional fields.
100- Use `str | None` syntax, not `Optional[str]`.
101- Enums for status fields: use Python `enum.StrEnum` (3.11+) or `str, Enum`.
102- Never reuse a request schema as a response schema. Separate them even if they look similar.
103 
104### Configuration
105 
106All config from environment variables via `pydantic-settings`. No `os.getenv()` scattered through code.
107 
108```python
109# app/config.py
110from pydantic_settings import BaseSettings
111 
112class Settings(BaseSettings):
113    model_config = ConfigDict(env_file=".env", env_file_encoding="utf-8")
114 
115    database_url: str
116    redis_url: str = "redis://localhost:6379/0"
117    secret_key: str
118    debug: bool = False
119    allowed_origins: list[str] = ["http://localhost:3000"]
120```
121 
122Settings instance created once in `app/main.py` or `app/dependencies.py`, injected via `Depends`. Never hardcode secrets. `.env.example` documents required vars.
123 
124### SQLAlchemy 2.0
125 
126Use SQLAlchemy 2.0 style throughout. No legacy 1.x patterns.
127 
128```python
129# app/models/order.py
130from sqlalchemy import ForeignKey, String
131from sqlalchemy.orm import Mapped, mapped_column, relationship
132from app.models.base import Base, TimestampMixin, UUIDMixin
133 
134class Order(UUIDMixin, TimestampMixin, Base):
135    __tablename__ = "orders"
136 
137    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
138    status: Mapped[str] = mapped_column(String(20), default="pending")
139    total_cents: Mapped[int] = mapped_column(default=0)
140 
141    user: Mapped["User"] = relationship(back_populates="orders")
142    items: Mapped[list["OrderItem"]] = relationship(back_populates="order", cascade="all, delete-orphan")
143```
144 
145**Rules:**
146- Use `Mapped[]` type annotations with `mapped_column()`. Never use `Column()` directly.
147- Base mixin for UUID primary keys and timestamps — define once in `app/models/base.py`.
148- Relationships always have `back_populates`, never `backref`.
149- Use async sessions (`AsyncSession`) if the project uses `asyncpg`. Use sync sessions if it uses `psycopg2`.
150 
151### Database Sessions & Dependencies
152 
153```python
154# app/dependencies.py
155async def get_db() -> AsyncGenerator[AsyncSession, None]:
156    async with async_session_maker() as session:
157        try:
158            yield session
159            await session.commit()
160        except Exception:
161            await session.rollback()
162            raise
163```
164 
165One session per request via `Depends(get_db)`. Commit at the dependency level, not in services. Services receive the session as an argument.
166 
167### Service Layer
168 
169Services are plain functions (or thin classes) that contain all business logic. They never import FastAPI.
170 
171```python
172# app/services/order.py
173from sqlalchemy import select
174from sqlalchemy.ext.asyncio import AsyncSession
175 
176async def create_order(
177    *,
178    db: AsyncSession,
179    user_id: uuid.UUID,
180    items: list[OrderItemCreate],
181    shipping_address_id: uuid.UUID,
182) -> Order:
183    address = await db.get(ShippingAddress, shipping_address_id)
184    if not address or address.user_id != user_id:
185        raise AddressNotFoundError(shipping_address_id)
186 
187    order = Order(user_id=user_id, status="pending")
188    for item in items:
189        product = await db.get(Product, item.product_id)
190        if not product or product.stock < item.quantity:
191            raise InsufficientStockError(item.product_id)
192        product.stock -= item.quantity
193        order.items.append(OrderItem(product_id=product.id, quantity=item.quantity, price_cents=product.price_cents))
194 
195    order.total_cents = sum(i.price_cents * i.quantity for i in order.items)
196    db.add(order)
197    await db.flush()
198    return order
199```
200 
201**Rules:**
202- Use keyword-only arguments (`*`) to force named calls.
203- Services raise domain exceptions (`OrderNotFoundError`, `InsufficientStockError`) — never `HTTPException`.
204- Services receive `AsyncSession` (or `Session`), not request objects.
205- Use `db.flush()` (not `db.commit()`) — the dependency commits.
206 
207### Route Handlers
208 
209```python
210# app/routers/order.py
211from fastapi import APIRouter, Depends, status
212 
213router = APIRouter(prefix="/orders", tags=["orders"])
214 
215@router.post("/", response_model=OrderResponse, status_code=status.HTTP_201_CREATED)
216async def create_order_endpoint(
217    body: OrderCreate,
218    db: AsyncSession = Depends(get_db),
219    current_user: User = Depends(get_current_user),
220) -> Order:
221    return await create_order(
222        db=db, user_id=current_user.id, items=body.items, shipping_address_id=body.shipping_address_id
223    )
224```
225 
226**Rules:**
227- Handlers are thin: parse request → call service → return response. No business logic.
228- Always use `response_model` for type safety and automatic serialization.
229- Always set explicit `status_code` for non-200 responses.
230- Domain exceptions are caught by exception handlers (registered in `app/main.py`), not by try/except in routes.
231 
232### Exception Handling
233 
234Define domain exceptions in `app/exceptions.py` (`NotFoundError`, `ConflictError`, etc.) inheriting from a base `DomainError`. Register exception handlers in `main.py` that map domain errors to HTTP status codes with a consistent error envelope: `{"error": {"code": "not_found", "message": "..."}}`. Never raise raw `HTTPException` from services.
235 
236### Alembic Migrations
237 
238- Every model change requires a migration. Run `alembic revision --autogenerate -m "description"` and **inspect the generated file** — autogenerate misses renames (shows as drop + add), custom types, and data migrations.
239- One migration per logical change. Never edit a migration applied in production — create a new one.
240- Test migrations: `alembic upgrade head && alembic downgrade base && alembic upgrade head` should work cleanly.
241 
242## Testing
243 
244### Setup
245 
246```toml
247# pyproject.toml
248[tool.pytest.ini_options]
249asyncio_mode = "auto"
250addopts = "-v --tb=short -x --strict-markers"
251markers = ["slow: marks tests as slow"]
252```
253 
254### Fixtures
255 
256```python
257# tests/conftest.py
258import pytest
259from httpx import ASGITransport, AsyncClient
260from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
261 
262@pytest.fixture
263async def db_session():
264    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
265    async with engine.begin() as conn:
266        await conn.run_sync(Base.metadata.create_all)
267    session_maker = async_sessionmaker(engine, expire_on_commit=False)
268    async with session_maker() as session:
269        yield session
270 
271@pytest.fixture
272async def client(db_session):
273    app.dependency_overrides[get_db] = lambda: db_session
274    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
275        yield c
276    app.dependency_overrides.clear()
277```
278 
279### Test Patterns
280 
281```python
282async def test_create_order_success(client, db_session, user_factory, product_factory):
283    user = await user_factory(db_session)
284    product = await product_factory(db_session, stock=10, price_cents=1500)
285 
286    response = await client.post("/orders/", json={
287        "items": [{"product_id": str(product.id), "quantity": 2}],
288        "shipping_address_id": str(user.default_address_id),
289    }, headers={"Authorization": f"Bearer {create_token(user)}"})
290 
291    assert response.status_code == 201
292    assert response.json()["total_cents"] == 3000
293```
294 
295- Use `httpx.AsyncClient` with `ASGITransport` — not `TestClient` (which is sync).
296- Override dependencies via `app.dependency_overrides` in fixtures.
297- Test services directly for business logic. Test routes for HTTP-level behavior.
298- Use `polyfactory` or `factory-boy` for test data. Mock external services — never call real APIs in tests.
299 
300## Exploring the Codebase
301 
302Before making changes, orient yourself:
303 
304```bash
305# Find the FastAPI app instance and lifespan setup
306rg "FastAPI\(" app/main.py
307 
308# Find all registered routers
309rg "include_router" app/main.py
310 
311# Find all route handlers for a resource
312rg "@router\.(get|post|put|patch|delete)" app/routers/order.py
313 
314# Find a model definition
315rg "class Order" app/models/
316 
317# Find which services exist for a domain
318ls app/services/
319 
320# Find all Depends() injections for a dependency
321rg "Depends\(get_current_user\)" app/routers/
322 
323# Check Alembic migration history
324ls -la alembic/versions/
325 
326# Find all domain exceptions
327rg "class.*Error.*Exception\)|class.*Error.*DomainError" app/
328 
329# See what env vars the app requires
330cat app/config.py
331 
332# Check existing test patterns
333head -50 tests/conftest.py
334```
335 
336## When to Ask vs. Proceed
337 
338**Just do it:**
339- Adding a new route + service that follows an existing pattern
340- Writing or updating tests
341- Fixing lint or type errors
342- Adding Pydantic schemas for a new endpoint
343- Creating an Alembic migration for a model change
344- Adding input validation to an existing schema
345 
346**Ask first:**
347- Adding a new dependency to `pyproject.toml`
348- Changing the database session strategy (sync ↔ async)
349- Modifying authentication/authorization middleware
350- Changing the project structure or package layout
351- Switching ORMs, migration tools, or the ASGI server
352- Modifying shared base classes (`Base`, `TimestampMixin`)
353- Any change to `alembic/env.py`
354 
355## Git Workflow
356 
357- Branch from `main`. Name: `feat/<thing>`, `fix/<thing>`, `chore/<thing>`.
358- Write clear commit messages: imperative mood, reference ticket if one exists.
359- Before pushing, always run:
360  ```bash
361  ruff check .
362  ruff format . --check
363  pytest
364  ```
365- If tests fail, fix them before committing. Never skip or `@pytest.mark.skip` without a reason.
366- One logical change per commit. Separate migrations from application code commits.
367 
368## Common Pitfalls
369 
370- **Pydantic v1 patterns.** Never use `class Config: orm_mode = True`, `@validator`, or `Field(None, ...)` with v1 syntax. Use `model_config = ConfigDict(from_attributes=True)`, `@field_validator`, and `str | None = None`.
371- **Forgetting `await`.** Missed awaits on async DB calls fail silently or return coroutines. If a test returns unexpected types, check for missing `await`.
372- **N+1 queries.** Use `selectinload()` or `joinedload()` for relationships accessed in loops. Check the SQL log in tests.
373- **Committing in services.** Services call `db.flush()`, not `db.commit()`. The session dependency handles the commit.
374- **Mutable defaults.** Never use `default=[]` or `default={}` in function signatures or Pydantic fields. Use `default_factory=list`.
375- **Importing between routers.** Routers should be independent. Shared logic goes in services; shared types go in schemas.
376- **Background tasks and DB sessions.** A `BackgroundTask` runs after the response, when the request session is closed. If it needs DB access, create a new session inside the task.
377