Enforces FastAPI patterns with DDD and Clean Architecture: endpoint structure, UoW dependency injection, domain exception handling, pure ASGI middleware, testing. Activates when working with FastAPI routers, endpoints, or middleware.
Scanned 5/28/2026
Install via CLI
Skills Directoryopenskills install AppVerk/av-marketplace---
name: fastapi-patterns
description: Enforces FastAPI patterns with DDD and Clean Architecture: endpoint structure, UoW dependency injection, domain exception handling, pure ASGI middleware, testing. Activates when working with FastAPI routers, endpoints, or middleware.
allowed-tools: Read, Grep, Glob, Bash(ruff:*), Bash(mypy:*), Bash(basedpyright:*), Bash(make:*), Bash(uv:*), Bash(python:*), Bash(pytest:*), Bash(uvicorn:*)
---
# FastAPI Patterns (DDD / Clean Architecture)
<HARD-RULES>
These rules are NON-NEGOTIABLE. Violating any of them is a bug.
**Architecture & Layers:**
- NEVER define endpoints directly on the `FastAPI()` app instance — ALWAYS use `APIRouter` and include it via `app.include_router()`
- NEVER inject `AsyncSession` into endpoints — ALWAYS use `UnitOfWork` Protocol dependency from the domain layer
- NEVER raise `HTTPException` in services or domain layer — services raise domain exceptions, the presentation layer maps them to HTTP responses via global exception handlers
- NEVER return domain entities directly from endpoints — ALWAYS map through response schemas with `from_domain()` classmethod
- NEVER import SQLAlchemy or infrastructure types in routers — only domain and schema types
**Endpoints:**
- NEVER use synchronous `def` for endpoints that perform I/O — ALWAYS use `async def`
- NEVER return raw dicts from endpoints — ALWAYS use a typed return annotation with a Pydantic model
- NEVER catch generic `Exception` in endpoints — use domain exception handlers or raise `HTTPException` for HTTP-level concerns (auth, validation)
- ALWAYS use `Annotated[T, Depends(...)]` for dependency injection — NEVER pass `Depends()` as a default parameter value
- ALWAYS declare explicit `status_code` on mutating endpoints (`POST`, `PUT`, `PATCH`, `DELETE`)
**Schemas & Types:**
- NEVER use separate Create/Read schemas named differently per endpoint — ALWAYS follow the `ItemCreate`, `ItemUpdate`, `ItemResponse` naming convention
- ALWAYS use PEP 695 type parameter syntax (`class Foo[T]`, `type Alias = ...`) — NEVER use `TypeVar` / `Generic[T]` / `TypeAlias`
**Middleware & Config:**
- NEVER use `BaseHTTPMiddleware` — ALWAYS use pure ASGI middleware (BaseHTTPMiddleware has memory leaks, breaks contextvars, and will be removed in Starlette 1.0)
- NEVER hardcode CORS origins in source code — ALWAYS load them from settings/environment
- NEVER use `@app.on_event("startup")` or `@app.on_event("shutdown")` — ALWAYS use the `lifespan` async context manager
**Quality:**
- ALWAYS run the project's typecheck and test commands after any endpoint change
</HARD-RULES>
These are the FastAPI patterns for building APIs in AppVerk projects, following **Domain-Driven Design** and **Clean Architecture**. All patterns target **Python 3.13+**, **FastAPI 0.128+**, and **Pydantic v2.7+**.
## Architecture Overview
```
┌──────────────────────────────────────────────────────┐
│ Presentation Layer (FastAPI routers, schemas, │
│ exception handlers, middleware) │
│ Depends on: Application Layer, Domain Layer │
├──────────────────────────────────────────────────────┤
│ Application Layer (Services / Use Cases) │
│ Depends on: Domain Layer only │
│ Uses: UnitOfWork Protocol, Repository Protocol │
├──────────────────────────────────────────────────────┤
│ Domain Layer (Entities, Value Objects, │
│ Repository Protocols, UoW Protocol, │
│ Domain Exceptions) │
│ Depends on: NOTHING │
├──────────────────────────────────────────────────────┤
│ Infrastructure Layer (ORM models, SA repos, │
│ SA UoW, Data Mappers, Alembic) │
│ Depends on: Domain Layer + SQLAlchemy │
└──────────────────────────────────────────────────────┘
```
**FastAPI lives in the Presentation Layer.** It receives HTTP requests, delegates to services, maps results to responses, and translates domain exceptions to HTTP status codes. It NEVER touches infrastructure directly.
**Dependency direction:** ALWAYS inward. Routers depend on services (application layer) and schemas (presentation). Services depend on domain Protocols. Infrastructure implements those Protocols.
### Directory Layout
```
app/
api/ # Presentation layer
routes/ # One file per domain module
markets.py
users.py
dependencies.py # DI: UoW, services, auth
exception_handlers.py # Domain exception → HTTP mapping
schemas/ # Pydantic request/response models
market.py
user.py
common.py # PaginatedResponse, ErrorResponse
domain/
entities/ # Pure dataclasses / Pydantic models
value_objects/ # Immutable value types
repositories/ # Protocol interfaces
exceptions.py # Domain exception hierarchy
unit_of_work.py # UoW Protocol
services/ # Application layer (use cases)
infrastructure/
persistence/
models/ # SQLAlchemy ORM models (*ORM)
mappers/ # Domain ↔ ORM conversion
repositories/ # Repository implementations
unit_of_work.py # SQLAlchemy UoW
database.py # Engine, session factory
middleware/ # Pure ASGI middleware
main.py # App factory, lifespan, router registration
config.py # Settings (pydantic-settings)
```
## Router Structure
One router per domain module. Each router lives in its own file under `api/routes/`.
```python
# app/api/routes/markets.py
from fastapi import APIRouter
router = APIRouter(
prefix="/markets",
tags=["markets"],
)
@router.get("/")
async def list_markets(
filters: Annotated[MarketFilters, Query()],
service: MarketServiceDep,
) -> PaginatedResponse[MarketResponse]:
...
@router.get("/{market_id}")
async def get_market(
market_id: UUID,
service: MarketServiceDep,
) -> MarketResponse:
market = await service.get_by_id(market_id)
return MarketResponse.from_domain(market)
@router.post("/", status_code=201)
async def create_market(
body: MarketCreate,
service: MarketServiceDep,
user: CurrentUser,
) -> MarketResponse:
market = await service.create_market(body, user_id=user.id)
return MarketResponse.from_domain(market)
```
Register all routers in the app factory:
```python
# app/main.py
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI
from app.api.exception_handlers import register_exception_handlers
from app.api.routes import markets, users, trades
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# startup: open DB pool, warm caches, etc.
yield
# shutdown: close DB pool, flush buffers, etc.
def create_app() -> FastAPI:
app = FastAPI(title="AV Marketplace", lifespan=lifespan)
register_exception_handlers(app)
app.include_router(markets.router, prefix="/api/v1")
app.include_router(users.router, prefix="/api/v1")
app.include_router(trades.router, prefix="/api/v1")
return app
```
### Conventions
- **Prefix**: set on the router (`prefix="/markets"`), not at include time, unless you need a version prefix like `/api/v1`.
- **Tags**: one tag per router matching the domain name. Used for OpenAPI docs grouping.
- **File layout**: `app/api/routes/<domain>.py` with a module-level `router` variable.
- **Return type as response model**: use typed return annotations (e.g., `-> MarketResponse`) instead of `response_model=` parameter. Use `response_model` only when you need field filtering (`response_model_include`, `response_model_exclude`).
## Dependency Injection — UoW & Services
Use `Annotated` types with `Depends()` for all injected dependencies. Define reusable type aliases. **Endpoints receive services, not infrastructure types.**
### UoW Dependency
The Unit of Work dependency bridges infrastructure and presentation. This is the **only place** where infrastructure types appear in the DI graph:
```python
# app/api/dependencies.py
from typing import Annotated
from collections.abc import AsyncIterator
from fastapi import Depends
from app.domain.unit_of_work import UnitOfWork
from app.infrastructure.persistence.database import async_session_factory
from app.infrastructure.persistence.unit_of_work import SqlAlchemyUnitOfWork
async def get_unit_of_work() -> AsyncIterator[SqlAlchemyUnitOfWork]:
uow = SqlAlchemyUnitOfWork(async_session_factory)
yield uow
UoW = Annotated[UnitOfWork, Depends(get_unit_of_work)]
```
**Key point:** the type alias `UoW` is typed as `UnitOfWork` (the domain Protocol), not `SqlAlchemyUnitOfWork`. Endpoints see the abstraction.
### Service Dependencies
Services are composed from UoW via nested dependencies. FastAPI resolves the graph automatically:
```python
# app/api/dependencies.py
from app.services.market import MarketService
from app.services.user import UserService
async def get_market_service(uow: UoW) -> MarketService:
return MarketService(uow)
async def get_user_service(uow: UoW) -> UserService:
return UserService(uow)
MarketServiceDep = Annotated[MarketService, Depends(get_market_service)]
UserServiceDep = Annotated[UserService, Depends(get_user_service)]
```
### Authentication
```python
# app/api/dependencies.py
from fastapi import HTTPException, Security, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from app.domain.entities.user import User
from app.services.auth import AuthService
security = HTTPBearer()
async def get_auth_service(uow: UoW) -> AuthService:
return AuthService(uow)
async def get_current_user(
credentials: Annotated[HTTPAuthorizationCredentials, Security(security)],
auth_service: Annotated[AuthService, Depends(get_auth_service)],
) -> User:
user = await auth_service.verify_token(credentials.credentials)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
)
return user
CurrentUser = Annotated[User, Depends(get_current_user)]
```
**Notes:**
- Use `Security()` instead of `Depends()` for security dependencies — this generates proper OpenAPI security scheme documentation and enables OAuth2 scopes.
- `HTTPBearer` is for pure Bearer token auth (JWT/API key obtained externally).
- `HTTPException` is acceptable here because authentication is a **presentation-layer concern** — the service (`AuthService`) returns `None` for invalid tokens, the DI function translates that to HTTP 401.
### OAuth2 Password Flow (Swagger UI Login)
When your API has a username/password login endpoint and you want the Swagger UI "Authorize" button to work with credentials:
```python
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
@router.post("/auth/token")
async def login(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
auth_service: Annotated[AuthService, Depends(get_auth_service)],
) -> TokenResponse:
token = await auth_service.authenticate(form_data.username, form_data.password)
return TokenResponse(access_token=token, token_type="bearer")
```
### Using Dependencies in Endpoints
```python
# app/api/routes/markets.py
from app.api.dependencies import CurrentUser, MarketServiceDep
@router.post("/", status_code=201)
async def create_market(
body: MarketCreate,
service: MarketServiceDep,
user: CurrentUser,
) -> MarketResponse:
market = await service.create_market(body, user_id=user.id)
return MarketResponse.from_domain(market)
```
**What happens here:**
1. FastAPI resolves `MarketServiceDep` → calls `get_market_service(uow)` → calls `get_unit_of_work()`
2. Service receives `UnitOfWork` Protocol — no infrastructure types in the endpoint
3. Service returns a domain entity (`Market` dataclass)
4. Endpoint maps it to `MarketResponse` via `from_domain()`
5. If service raises a domain exception (e.g., `EntityNotFoundError`), the global handler maps it to HTTP 404
## Request/Response Schemas
Separate Pydantic models for input and output. **Response schemas map from domain entities, not ORM models.**
```python
# app/schemas/market.py
from datetime import datetime
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field
from app.domain.entities.market import Market
class MarketCreate(BaseModel):
"""Input schema for creating a market."""
name: str = Field(min_length=1, max_length=255)
description: str = Field(min_length=1, max_length=5000)
end_date: datetime
class MarketUpdate(BaseModel):
"""Input schema for partial market update."""
name: str | None = Field(default=None, min_length=1, max_length=255)
description: str | None = Field(default=None, max_length=5000)
end_date: datetime | None = None
class MarketResponse(BaseModel):
"""Output schema returned to clients."""
model_config = ConfigDict(from_attributes=True)
id: UUID
name: str
description: str
slug: str
status: str
end_date: datetime
created_at: datetime
updated_at: datetime
@classmethod
def from_domain(cls, entity: Market) -> "MarketResponse":
"""Map a domain entity to a response schema."""
return cls.model_validate(entity, from_attributes=True)
```
### Conventions
- `<Entity>Create` — required fields for creation. Use `Field()` constraints.
- `<Entity>Update` — all fields optional for partial updates.
- `<Entity>Response` — full representation returned to clients. `from_attributes=True` works with both dataclass entities and ORM models. The `from_domain()` classmethod documents the mapping intent.
- For list endpoints returning paginated data, use a generic wrapper:
```python
# app/schemas/common.py
from pydantic import BaseModel
class PaginatedResponse[T](BaseModel):
data: list[T]
total: int
limit: int
offset: int
```
## Domain Exception Handling
Services raise **domain exceptions**. The presentation layer maps them to HTTP responses via global exception handlers. **Endpoints never catch domain exceptions manually** — the handlers do it automatically.
### Domain Exception Hierarchy
```python
# app/domain/exceptions.py — ZERO external dependencies
class DomainError(Exception):
"""Base for all domain exceptions."""
class EntityNotFoundError(DomainError):
"""Raised when a requested entity does not exist."""
class PermissionDeniedError(DomainError):
"""Raised when the user lacks permission for the operation."""
class BusinessRuleViolationError(DomainError):
"""Raised when a business rule is violated (e.g., market already closed)."""
class DuplicateEntityError(DomainError):
"""Raised when creating an entity that already exists (e.g., duplicate slug)."""
```
### Global Exception Handlers
```python
# app/api/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.domain.exceptions import (
BusinessRuleViolationError,
DomainError,
DuplicateEntityError,
EntityNotFoundError,
PermissionDeniedError,
)
DOMAIN_EXCEPTION_MAP: dict[type[DomainError], int] = {
EntityNotFoundError: 404,
PermissionDeniedError: 403,
BusinessRuleViolationError: 422,
DuplicateEntityError: 409,
}
def register_exception_handlers(app: FastAPI) -> None:
"""Register domain exception → HTTP status code mappings."""
for exc_class, status_code in DOMAIN_EXCEPTION_MAP.items():
@app.exception_handler(exc_class)
async def _handler(
request: Request,
exc: DomainError,
sc: int = status_code,
) -> JSONResponse:
return JSONResponse(
status_code=sc,
content={
"detail": str(exc),
"error_code": type(exc).__name__,
},
)
```
### How It Flows
```
Endpoint → Service → raises EntityNotFoundError("Market 123 not found")
↓
Global exception handler catches it
↓
Returns JSONResponse(status_code=404, ...)
```
The endpoint code stays clean — no try/except, no HTTPException for business logic:
```python
@router.get("/{market_id}")
async def get_market(
market_id: UUID,
service: MarketServiceDep,
) -> MarketResponse:
# If market doesn't exist, service raises EntityNotFoundError
# Global handler converts it to 404 — no manual catching needed
market = await service.get_by_id(market_id)
return MarketResponse.from_domain(market)
```
### Standard Error Response Format
All error responses follow this shape:
```json
{
"detail": "Human-readable error message",
"error_code": "EntityNotFoundError"
}
```
### When to Use HTTPException Directly
`HTTPException` is still appropriate for **presentation-layer concerns** that don't involve domain logic:
- Authentication failures (401) — in DI functions, not services
- Request validation beyond Pydantic (rare)
- Rate limiting responses
## Query Parameter Models
Since FastAPI 0.115, you can group related query parameters into Pydantic models. This is the preferred pattern for endpoints with multiple filters.
```python
from fastapi import Query
from pydantic import BaseModel, ConfigDict, Field
from app.domain.value_objects.market_status import MarketStatus
class MarketFilters(BaseModel):
"""Query parameters for filtering markets."""
model_config = ConfigDict(extra="forbid")
status: MarketStatus | None = None
created_by_id: UUID | None = None
limit: int = Field(default=20, le=100, ge=1)
offset: int = Field(default=0, ge=0)
@router.get("/")
async def list_markets(
filters: Annotated[MarketFilters, Query()],
service: MarketServiceDep,
) -> PaginatedResponse[MarketResponse]:
markets, total = await service.list_markets(
status=filters.status,
created_by_id=filters.created_by_id,
offset=filters.offset,
limit=filters.limit,
)
return PaginatedResponse(
data=[MarketResponse.from_domain(m) for m in markets],
total=total,
limit=filters.limit,
offset=filters.offset,
)
```
### Benefits
- **Reusable**: the same filter model can be shared across related endpoints.
- **Validated**: `Field()` constraints are enforced automatically.
- **Strict**: `extra="forbid"` rejects unknown query parameters.
- **Type-safe**: IDE autocompletion on the `filters` object.
### Header and Cookie Models
The same pattern works for headers and cookies (also since 0.115):
```python
from fastapi import Cookie, Header
class TraceHeaders(BaseModel):
x_request_id: str | None = None
x_correlation_id: str | None = None
class SessionCookies(BaseModel):
session_token: str | None = None
@router.get("/debug")
async def debug_info(
headers: Annotated[TraceHeaders, Header()],
cookies: Annotated[SessionCookies, Cookie()],
) -> dict:
...
```
## Background Tasks
Use `BackgroundTasks` for lightweight fire-and-forget work that runs after the response is sent.
```python
from fastapi import BackgroundTasks
async def send_welcome_email(email: str) -> None:
# lightweight I/O, not CPU-bound
...
@router.post("/", status_code=201)
async def register_user(
body: UserCreate,
service: UserServiceDep,
user: CurrentUser,
background_tasks: BackgroundTasks,
) -> UserResponse:
created = await service.create_user(body, created_by=user.id)
background_tasks.add_task(send_welcome_email, created.email)
return UserResponse.from_domain(created)
```
### When to Use BackgroundTasks vs a Task Queue
| Scenario | Use |
|---|---|
| Send a notification email | `BackgroundTasks` |
| Write an audit log entry | `BackgroundTasks` |
| Generate a PDF report (seconds) | `BackgroundTasks` |
| Process a large file upload (minutes) | Celery / ARQ / TaskIQ |
| Run an ML inference pipeline | Celery / ARQ / TaskIQ |
| Any job that must survive a server restart | Celery / ARQ / TaskIQ |
Rule of thumb: if the task takes more than a few seconds or must be retried on failure, use an external task queue.
**Note:** Since FastAPI 0.118, the exit code of `yield` dependencies runs **after** the response is sent. This means cleanup logic (e.g., closing sessions) happens after background tasks finish.
## Lifespan Events
Use `@asynccontextmanager` lifespan for startup/shutdown logic. Never use the deprecated `on_event` decorators.
```python
# app/main.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy import text
from app.infrastructure.persistence.database import engine
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# --- Startup ---
# Verify DB connectivity (engine is lazy)
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
yield
# --- Shutdown ---
await engine.dispose()
```
### What Goes in Lifespan
- **Startup**: open connection pools (DB, Redis, HTTP clients), load configuration, warm caches.
- **Shutdown**: close connection pools, flush pending writes, release resources.
- **Never**: run long-lived background loops here — use a dedicated task runner.
## Middleware — Pure ASGI
**NEVER use `BaseHTTPMiddleware`.** It has known, unfixable issues:
- Buffers entire response in memory (no backpressure — OOM risk)
- Breaks `contextvars` propagation between middleware and endpoints
- Incompatible with `BackgroundTasks` (timeout issues, swallowed errors)
- Starlette plans to remove it in version 1.0
**ALWAYS use pure ASGI middleware** — it's how Starlette's own built-in middleware is implemented.
### CORS
Load allowed origins from settings. Never hardcode them.
```python
# app/main.py
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
def create_app() -> FastAPI:
app = FastAPI(title="AV Marketplace", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_allowed_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ... register exception handlers, include routers
return app
```
### Custom Middleware — Request ID Example
```python
# app/middleware/request_id.py
import uuid
from starlette.types import ASGIApp, Message, Receive, Scope, Send
class RequestIdMiddleware:
"""Injects X-Request-ID into request scope and response headers."""
def __init__(self, app: ASGIApp) -> None:
self.app = app
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return
# Extract from incoming header or generate new
headers = dict(scope.get("headers", []))
request_id = headers.get(b"x-request-id", b"").decode() or str(uuid.uuid4())
# Store in scope for access by endpoints via request.scope["state"]
scope.setdefault("state", {})["request_id"] = request_id
async def send_wrapper(message: Message) -> None:
if message["type"] == "http.response.start":
response_headers = list(message.get("headers", []))
response_headers.append((b"x-request-id", request_id.encode()))
message["headers"] = response_headers
await send(message)
await self.app(scope, receive, send_wrapper)
```
Register middleware in the app factory:
```python
from app.middleware.request_id import RequestIdMiddleware
app.add_middleware(RequestIdMiddleware)
```
### Custom Middleware — Timing Example
```python
# app/middleware/timing.py
import time
from starlette.types import ASGIApp, Message, Receive, Scope, Send
class TimingMiddleware:
"""Adds X-Process-Time header to responses."""
def __init__(self, app: ASGIApp) -> None:
self.app = app
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return
start = time.perf_counter()
async def send_wrapper(message: Message) -> None:
if message["type"] == "http.response.start":
elapsed = time.perf_counter() - start
headers = list(message.get("headers", []))
headers.append((b"x-process-time", f"{elapsed:.4f}".encode()))
message["headers"] = headers
await send(message)
await self.app(scope, receive, send_wrapper)
```
### Pure ASGI Middleware Pattern
The structure is always the same:
```python
class MyMiddleware:
def __init__(self, app: ASGIApp) -> None:
self.app = app
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return
# Before request: modify scope or receive
# ...
async def send_wrapper(message: Message) -> None:
# After response start: modify headers
# After response body: inspect/log body chunks
await send(message)
await self.app(scope, receive, send_wrapper)
```
**Key concepts:**
- `scope`: dict with request metadata (type, path, headers, state). Modify to pass data to endpoints.
- `receive`: async callable that returns request body chunks. Wrap to intercept/log body.
- `send`: async callable that sends response chunks. Wrap to modify response headers or body.
- Always pass through non-HTTP scopes (`websocket`, `lifespan`) with `await self.app(scope, receive, send)`.
### Middleware Ordering
Middleware executes in **reverse registration order** (last registered = first to execute). Register in this order:
```python
# 1. CORS — must be outermost (registered first, executes last)
app.add_middleware(CORSMiddleware, ...)
# 2. Request ID — inject early so all subsequent middleware and endpoints see it
app.add_middleware(RequestIdMiddleware)
# 3. Timing — measure after request ID is set
app.add_middleware(TimingMiddleware)
```
Execution flow for an incoming request:
```
Request → TimingMiddleware → RequestIdMiddleware → CORSMiddleware → Router → Endpoint
Response ← TimingMiddleware ← RequestIdMiddleware ← CORSMiddleware ← Router ← Endpoint
```
## Testing FastAPI
### pytest Configuration
Use modern `pytest-asyncio` config — no need for `@pytest.mark.asyncio` decorators or `event_loop` fixture:
```toml
# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
```
### Test Client Setup
Use `httpx.AsyncClient` with `ASGITransport` for all async endpoint tests:
```python
# tests/conftest.py
import pytest
from httpx import ASGITransport, AsyncClient
from app.main import create_app
from app.api.dependencies import get_unit_of_work, get_current_user
from tests.fakes import FakeUnitOfWork, make_fake_user
@pytest.fixture
def app():
app = create_app()
app.dependency_overrides[get_unit_of_work] = FakeUnitOfWork
app.dependency_overrides[get_current_user] = lambda: make_fake_user(role="admin")
yield app
app.dependency_overrides.clear()
@pytest.fixture
async def client(app):
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
```
### Overriding Dependencies
Override the **UoW dependency** (not the session). This keeps tests consistent with the DDD architecture:
```python
# tests/fakes.py
from uuid import UUID
from app.domain.entities.market import Market
from app.domain.entities.user import User
class FakeMarketRepository:
def __init__(self) -> None:
self.markets: dict[UUID, Market] = {}
async def get_by_id(self, entity_id: UUID) -> Market | None:
return self.markets.get(entity_id)
async def save(self, entity: Market) -> Market:
self.markets[entity.id] = entity
return entity
async def delete(self, entity: Market) -> None:
self.markets.pop(entity.id, None)
async def get_all(
self, *, offset: int = 0, limit: int = 100
) -> list[Market]:
items = list(self.markets.values())
return items[offset : offset + limit]
class FakeUnitOfWork:
def __init__(self) -> None:
self.markets = FakeMarketRepository()
self.committed = False
async def __aenter__(self) -> "FakeUnitOfWork":
return self
async def __aexit__(self, *args: object) -> None:
pass
async def commit(self) -> None:
self.committed = True
async def rollback(self) -> None:
pass
def make_fake_user(role: str = "user") -> User:
return User(
id=UUID("00000000-0000-0000-0000-000000000001"),
email="test@example.com",
role=role,
)
```
### Testing Endpoints
```python
# tests/functional/test_markets_api.py
from httpx import AsyncClient
class TestCreateMarket:
async def test_returns_201_with_valid_data(self, client: AsyncClient) -> None:
response = await client.post(
"/api/v1/markets/",
json={
"name": "Election 2024",
"description": "US Presidential Election",
"end_date": "2024-11-05T00:00:00Z",
},
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "Election 2024"
assert "id" in data
async def test_returns_422_with_missing_name(self, client: AsyncClient) -> None:
response = await client.post(
"/api/v1/markets/",
json={"description": "Test", "end_date": "2024-11-05T00:00:00Z"},
)
assert response.status_code == 422
class TestGetMarket:
async def test_returns_404_for_nonexistent_market(
self, client: AsyncClient
) -> None:
response = await client.get(
"/api/v1/markets/00000000-0000-0000-0000-000000000000"
)
assert response.status_code == 404
async def test_returns_market_data(self, client: AsyncClient) -> None:
# Create first, then get
create_response = await client.post(
"/api/v1/markets/",
json={
"name": "Test Market",
"description": "Test",
"end_date": "2024-12-31T00:00:00Z",
},
)
market_id = create_response.json()["id"]
response = await client.get(f"/api/v1/markets/{market_id}")
assert response.status_code == 200
assert response.json()["name"] == "Test Market"
class TestListMarkets:
async def test_returns_paginated_response(self, client: AsyncClient) -> None:
response = await client.get("/api/v1/markets/?limit=10&offset=0")
assert response.status_code == 200
data = response.json()
assert isinstance(data["data"], list)
assert "total" in data
```
### Testing with Lifespan Events
`AsyncClient` does NOT trigger lifespan events. Use synchronous `TestClient` in a `with` block for lifespan-dependent tests:
```python
from fastapi.testclient import TestClient
def test_health_check(app):
with TestClient(app) as client:
response = client.get("/health")
assert response.status_code == 200
```
### Testing Conventions
- Override `get_unit_of_work` with `FakeUnitOfWork` — never mock FastAPI or SQLAlchemy internals.
- Test the HTTP interface (status codes, response bodies, headers), not service internals.
- Each test class targets one endpoint or related group of endpoints.
- Always clear `dependency_overrides` after tests (use `yield` fixtures).
- No `@pytest.mark.asyncio` needed — `asyncio_mode = "auto"` handles it.
- See the `tdd-workflow` skill for full testing rules and Fake vs Mock guidance.
- See the `sqlalchemy-patterns` skill for `FakeUnitOfWork` and `FakeRepository` patterns.
No comments yet. Be the first to comment!
