From 52707b6917eaa42dd818823c58d431a90a09796c Mon Sep 17 00:00:00 2001 From: Yerassyl Date: Tue, 14 Apr 2026 21:00:34 +0500 Subject: [PATCH] Refresh examples for modern stack; promote AGENTS.md to canonical agent ruleset (#89) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Refresh examples for modern stack; promote AGENTS.md to canonical agent ruleset Replaces stale code in README.md (python-jose, async_asgi_testclient, json_encoders, contradictory `Field(ge=18, default=None)`, unused `model_validator` import) with current equivalents (PyJWT, httpx + ASGITransport, `@field_serializer`). Adds a top-of-file pointer to AGENTS.md, a tested-versions line, and inline notes for `Annotated[T, Depends(...)]` and SQLAlchemy 2.0 async without rewriting every example. Adds a `dependency_overrides` testing subsection and a BackgroundTasks vs. task-queue decision table. Rewrites AGENTS.md as a terse, machine-readable mirror: compatibility matrix at the top, Do/Don't blocks throughout, an explicit anti-patterns table covering common AI-agent mistakes, and a quick-reference table. Co-Authored-By: Claude Opus 4.6 (1M context) * Sync code-only fixes into README_ZH.md Mirrors the safe code-only changes from the parent commit into the Chinese translation: - python-jose -> PyJWT (import + exception class) in both dependency examples - async_asgi_testclient -> httpx.AsyncClient + ASGITransport (and fixes the previously broken example that mixed both imports) - Pydantic json_encoders -> @field_serializer for the CustomModel example - Field(ge=18, default=None) -> Field(ge=18) (the constraint and the default contradict each other) - Removes unused `model_validator` import No prose changes. Existing Chinese descriptions still accurately describe the new code. Sections newly added in README.md (BackgroundTasks vs task queue, dependency_overrides, the agent banner, the version line, and the Annotated/SQLAlchemy 2.0 inline notes) are not ported here because they require translation. Co-Authored-By: Claude Opus 4.6 (1M context) * Trim docs/changelog voice from refresh Pass over the refresh to keep the article in best-practices voice, not docs or changelog: - Drop the "Note on syntax" blockquote in README's Dependencies section. It hedged ("Both styles work. New code should prefer Annotated.") while every example below kept the default-arg form. Removing the callout returns the section to its original shape; a full Annotated migration can be its own PR. - Remove the "json_encoders is deprecated in Pydantic v2" inline explainer. The code change already says it. - Rewrite the databases blockquote into a single in-voice line ("For new projects, reach for SQLAlchemy 2.0's async API...") instead of an apologetic disclaimer. - Collapse the "Earlier versions of this article recommended async_asgi_testclient" blockquote into a direct sentence. - Drop the "this file wins — README is narrative, this is the spec" precedence line in AGENTS.md. The two files are peers; neither outranks the other. Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- AGENTS.md | 455 ++++++++++++++++++++++++++++++++++++--------------- README.md | 106 +++++++++--- README_ZH.md | 51 +++--- 3 files changed, 433 insertions(+), 179 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 5063726..1c15b2b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,160 +1,266 @@ # FastAPI Best Practices for AI Agents -This document provides guidelines for AI agents working on FastAPI projects. Follow these conventions when writing or modifying code. +A machine-readable companion to [README.md](./README.md) for AI coding agents +working in FastAPI projects. Same rules, restructured for fast pattern matching: +version pins, Do/Don't blocks, anti-patterns, and a quick-reference table. + +## Compatibility Matrix + +Pin to these versions or newer. Examples in this file assume them. + +| Dependency | Minimum | Notes | +|------------------|-----------|------------------------------------------------------| +| Python | 3.11 | Required for `StrEnum` and `X \| Y` union syntax | +| FastAPI | 0.115 | `Annotated[T, Depends(...)]` is the idiomatic form | +| Pydantic | 2.7 | v1 APIs (`json_encoders`, `.dict()`) are removed | +| pydantic-settings| 2.4 | Lives in a separate package since Pydantic v2 | +| SQLAlchemy | 2.0 | Use the async API (`AsyncSession`, `async_sessionmaker`) | +| Alembic | 1.13 | Async-aware migrations | +| httpx | 0.27 | Use `ASGITransport` for in-process tests | +| PyJWT | 2.9 | Use this, not the unmaintained `python-jose` | +| ruff | 0.6 | Replaces black, isort, autoflake | ## Project Structure -Organize code by domain, not by file type. +Organize by domain, not by file type. One package per bounded context. ``` src/ ├── {domain}/ # e.g., auth/, posts/, aws/ │ ├── router.py # API endpoints │ ├── schemas.py # Pydantic models -│ ├── models.py # Database models +│ ├── models.py # SQLAlchemy ORM models │ ├── service.py # Business logic │ ├── dependencies.py # Route dependencies -│ ├── config.py # Environment variables +│ ├── config.py # Domain-scoped BaseSettings │ ├── constants.py # Constants and error codes │ ├── exceptions.py # Domain-specific exceptions │ └── utils.py # Helper functions -├── config.py # Global configuration -├── models.py # Global models +├── config.py # Global BaseSettings +├── models.py # Shared Pydantic / ORM bases ├── exceptions.py # Global exceptions -├── database.py # Database connection -└── main.py # FastAPI app initialization +├── database.py # Async engine + session factory +└── main.py # FastAPI app + lifespan ``` -**Import Convention**: Use explicit module names when importing across domains: +**Cross-domain imports**: always use the explicit module name. Never `from src.auth import *`. + ```python from src.auth import constants as auth_constants from src.notifications import service as notification_service +from src.posts.constants import ErrorCode as PostsErrorCode ``` ## Async Routes -### Rules -- `async def` routes: Use ONLY non-blocking I/O (`await` calls) -- `def` routes (sync): Use for blocking I/O (runs in threadpool automatically) -- CPU-intensive work: Offload to Celery or multiprocessing +### Decision rule + +| Route does this | Use | +|----------------------------------------|-------------| +| `await`-able non-blocking I/O | `async def` | +| Blocking I/O (no async client exists) | `def` (sync, runs in threadpool) | +| Mix of both | `async def` + `run_in_threadpool` for the blocking part | +| CPU-bound work (>50 ms compute) | Offload to a worker process (Celery / RQ / Arq) | + +### Do / Don't -### Common Mistakes to Avoid ```python -# WRONG: Blocking call in async route +# DON'T — blocking call inside async route freezes the entire event loop @router.get("/bad") -async def bad_route(): - time.sleep(10) # Blocks entire event loop - return {"status": "done"} +async def bad(): + time.sleep(10) # blocks every request on this worker + return {"ok": True} -# CORRECT: Non-blocking in async route -@router.get("/good") -async def good_route(): - await asyncio.sleep(10) - return {"status": "done"} +# DO — sync route lets FastAPI run it in a threadpool +@router.get("/sync-ok") +def sync_ok(): + time.sleep(10) # blocks one threadpool worker, not the loop + return {"ok": True} -# CORRECT: Sync route for blocking operations -@router.get("/also-good") -def sync_route(): - time.sleep(10) # Runs in threadpool - return {"status": "done"} -``` +# DO — async route with awaitable sleep +@router.get("/async-ok") +async def async_ok(): + await asyncio.sleep(10) # yields control, loop keeps serving requests + return {"ok": True} -### Using Sync Libraries in Async Context -```python +# DO — async route that has to call a sync library from fastapi.concurrency import run_in_threadpool -@router.get("/") -async def call_sync_library(): - result = await run_in_threadpool(sync_client.make_request, data=my_data) +@router.get("/wrap") +async def wrap(): + result = await run_in_threadpool(legacy_sync_client.fetch, "id") return result ``` +### Threadpool caveats +- Default Starlette threadpool size is 40. Saturating it slows every sync route. +- Threads cost more than coroutines. Don't use sync routes "just because." + ## Pydantic -### Use Built-in Validators +### Use built-in validators ```python -from pydantic import BaseModel, EmailStr, Field +from enum import StrEnum +from pydantic import AnyUrl, BaseModel, EmailStr, Field + + +class MusicBand(StrEnum): + AEROSMITH = "AEROSMITH" + QUEEN = "QUEEN" + ACDC = "AC/DC" + class UserCreate(BaseModel): - username: str = Field(min_length=1, max_length=128, pattern="^[A-Za-z0-9-_]+$") + first_name: str = Field(min_length=1, max_length=128) + username: str = Field(min_length=1, max_length=128, pattern=r"^[A-Za-z0-9_-]+$") email: EmailStr - age: int = Field(ge=18) + age: int = Field(ge=18) # required, must be >= 18 + favorite_band: MusicBand | None = None + website: AnyUrl | None = None ``` -### Custom Base Model -Create a shared base model for consistent serialization: +> **Don't** write `Field(ge=18, default=None)`. The constraint and the default contradict +> each other. Decide: required (`Field(ge=18)`) or optional (`int | None = Field(default=None, ge=18)`). + +### Custom base model — modern serialization + +`json_encoders` is deprecated in Pydantic v2. Use `@field_serializer` for per-field rules, +or annotate a custom type with `PlainSerializer`. + ```python -from pydantic import BaseModel, ConfigDict +from datetime import datetime +from zoneinfo import ZoneInfo +from pydantic import BaseModel, ConfigDict, field_serializer + class CustomModel(BaseModel): - model_config = ConfigDict( - json_encoders={datetime: datetime_to_gmt_str}, - populate_by_name=True, - ) + model_config = ConfigDict(populate_by_name=True) + + @field_serializer("*", when_used="json", check_fields=False) + def _serialize_datetimes(self, value): + if isinstance(value, datetime): + if value.tzinfo is None: + value = value.replace(tzinfo=ZoneInfo("UTC")) + return value.strftime("%Y-%m-%dT%H:%M:%S%z") + return value ``` -### Split BaseSettings by Domain +### Split BaseSettings by domain + +`pydantic-settings` is its own package since Pydantic v2. + ```python # src/auth/config.py +from datetime import timedelta +from pydantic_settings import BaseSettings, SettingsConfigDict + + class AuthConfig(BaseSettings): + model_config = SettingsConfigDict(env_prefix="AUTH_", env_file=".env", extra="ignore") + JWT_ALG: str JWT_SECRET: str - JWT_EXP: int = 5 + JWT_EXP_MINUTES: int = 5 + REFRESH_TOKEN_KEY: str + REFRESH_TOKEN_EXP: timedelta = timedelta(days=30) + SECURE_COOKIES: bool = True + auth_settings = AuthConfig() ``` ## Dependencies -### Use for Validation, Not Just DI +### Use Annotated, not default-arg `Depends(...)` + +`Annotated[T, Depends(...)]` is the idiomatic form since FastAPI 0.95 and avoids +gotchas with default values. + ```python -async def valid_post_id(post_id: UUID4) -> dict[str, Any]: +# DO — modern Annotated form +from typing import Annotated +from fastapi import Depends + +PostDep = Annotated[dict, Depends(valid_post_id)] + +@router.get("/posts/{post_id}") +async def get_post(post: PostDep): + return post + +# Avoid — default-argument form (still works, but legacy) +@router.get("/posts/{post_id}") +async def get_post(post: dict = Depends(valid_post_id)): + return post +``` + +### Validate inside dependencies (not just inject) +```python +async def valid_post_id(post_id: UUID4) -> dict: post = await service.get_by_id(post_id) if not post: raise PostNotFound() return post - -@router.get("/posts/{post_id}") -async def get_post(post: dict[str, Any] = Depends(valid_post_id)): - return post ``` -### Chain Dependencies +### Chain dependencies for reuse ```python async def valid_owned_post( - post: dict[str, Any] = Depends(valid_post_id), - token_data: dict[str, Any] = Depends(parse_jwt_data), -) -> dict[str, Any]: + post: Annotated[dict, Depends(valid_post_id)], + token_data: Annotated[dict, Depends(parse_jwt_data)], +) -> dict: if post["creator_id"] != token_data["user_id"]: raise UserNotOwner() return post ``` -### Key Rules -- Dependencies are cached per request (same dependency called multiple times = one execution) -- Prefer `async` dependencies to avoid threadpool overhead -- Use consistent path variable names to enable dependency reuse +### Rules +- Dependencies are **cached per request**. Same `Depends(x)` called 5 times in one request → `x` runs once. +- Prefer `async def` dependencies. Sync deps run in the threadpool — wasted overhead for small CPU-only checks. +- Use **the same path-variable name** across endpoints when you want to share a dependency (e.g. `profile_id` in both `/profiles/{profile_id}` and `/creators/{profile_id}`). -## REST Conventions +## Authentication — JWT + +Use **`PyJWT`**, not `python-jose` (unmaintained). -Use consistent path variable names for dependency reuse: ```python -# Both use profile_id, enabling shared valid_profile_id dependency -GET /profiles/{profile_id} -GET /creators/{profile_id} +import jwt # PyJWT +from jwt.exceptions import InvalidTokenError + +def decode_token(token: str) -> dict: + try: + return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALG]) + except InvalidTokenError as exc: + raise InvalidCredentials() from exc ``` -## Database +## Database — SQLAlchemy 2.0 async -### Naming Conventions -- Use `lower_case_snake` format -- Singular table names: `post`, `user`, `post_like` -- Group related tables with prefix: `payment_account`, `payment_bill` -- DateTime suffix: `_at` (e.g., `created_at`) -- Date suffix: `_date` (e.g., `birth_date`) +Prefer SQLAlchemy 2.0's async API. `encode/databases` is in maintenance mode — don't pick it for new projects. -### Set Explicit Index Names ```python +# src/database.py +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +engine = create_async_engine(str(settings.DATABASE_URL), pool_pre_ping=True) +SessionFactory = async_sessionmaker(engine, expire_on_commit=False) + + +async def get_db() -> AsyncSession: + async with SessionFactory() as session: + yield session +``` + +### Naming conventions +- `lower_case_snake` +- Singular tables: `post`, `user`, `post_like` +- Group with prefix: `payment_account`, `payment_bill` +- `_at` suffix for `datetime`, `_date` suffix for `date` +- Use the same FK column name everywhere it appears (`profile_id`, not `user_id` in some tables and `profile_id` in others) + +### Index naming convention +```python +from sqlalchemy import MetaData + POSTGRES_INDEXES_NAMING_CONVENTION = { "ix": "%(column_0_label)s_idx", "uq": "%(table_name)s_%(column_0_name)s_key", @@ -162,89 +268,180 @@ POSTGRES_INDEXES_NAMING_CONVENTION = { "fk": "%(table_name)s_%(column_0_name)s_fkey", "pk": "%(table_name)s_pkey", } +metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION) ``` -### SQL-First Approach -Prefer database-level operations for: -- Complex joins -- Data aggregation -- Building nested JSON responses +### SQL-first, Pydantic-second +- Do joins, aggregation, and JSON shaping in SQL — Postgres is faster than CPython at this. +- Hydrate the result into Pydantic only for response validation, not for transformation. -## Migrations (Alembic) +## Background work — BackgroundTasks vs Celery -- Keep migrations static and reversible -- Use descriptive file names: `2022-08-24_post_content_idx.py` -- Configure in alembic.ini: - ```ini - file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s - ``` +| Use BackgroundTasks when… | Use Celery / Arq / RQ when… | +|------------------------------------------|--------------------------------------------| +| Task is < 1 second | Task takes seconds to minutes | +| Failure can be silently dropped | You need retries, dead-letter, or visibility| +| Task is in-process (send email, log row) | Task is CPU-heavy or needs a separate pool | +| You don't need scheduling | You need cron, ETA, or rate limiting | -## API Documentation - -### Hide Docs in Production ```python -SHOW_DOCS_ENVIRONMENT = ("local", "staging") +from fastapi import BackgroundTasks -app_configs = {"title": "My API"} -if ENVIRONMENT not in SHOW_DOCS_ENVIRONMENT: - app_configs["openapi_url"] = None - -app = FastAPI(**app_configs) +@router.post("/signup") +async def signup(data: SignupIn, bg: BackgroundTasks): + user = await service.create_user(data) + bg.add_task(send_welcome_email, user.email) # fire-and-forget, in-process + return user ``` -### Document Endpoints Properly -```python -@router.post( - "/endpoints", - response_model=DefaultResponseModel, - status_code=status.HTTP_201_CREATED, - description="Description of the endpoint", - tags=["Category"], - responses={ - status.HTTP_201_CREATED: {"model": CreatedResponse}, - status.HTTP_400_BAD_REQUEST: {"model": ErrorResponse}, - }, -) -``` +> BackgroundTasks run **after the response is sent, in the same worker process**. If the +> worker dies, the task is lost. There is no retry. Don't use them for anything you'd +> page on. ## Testing -Use async test client from the start: +### Async client from day one ```python import pytest from httpx import AsyncClient, ASGITransport +from src.main import app + + @pytest.fixture async def client(): - async with AsyncClient( - transport=ASGITransport(app=app), - base_url="http://test" - ) as client: - yield client + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as ac: + yield ac + @pytest.mark.asyncio -async def test_endpoint(client: AsyncClient): - resp = await client.post("/posts") +async def test_create_post(client: AsyncClient): + resp = await client.post("/posts", json={"title": "hi"}) assert resp.status_code == 201 ``` +> **Don't** use `async_asgi_testclient` — it's unmaintained. The example above (httpx + +> `ASGITransport`) is the supported path. + +### Override dependencies in tests +Don't monkeypatch internals. Use FastAPI's built-in `dependency_overrides`. + +```python +from src.auth.dependencies import parse_jwt_data +from src.main import app + + +def fake_user(): + return {"user_id": "00000000-0000-0000-0000-000000000001"} + + +@pytest.fixture(autouse=True) +def _override_auth(): + app.dependency_overrides[parse_jwt_data] = fake_user + yield + app.dependency_overrides.clear() +``` + +## Migrations (Alembic) + +- Migrations must be static and reversible. +- Use the async template: `alembic init -t async migrations` +- Descriptive filenames: + ```ini + # alembic.ini + file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s + ``` + → `2026-04-14_add_post_content_idx.py` + +## API documentation + +### Hide docs outside selected envs +```python +from fastapi import FastAPI +from src.config import settings + +SHOW_DOCS_IN = {"local", "staging"} +app_kwargs = {"title": "My API"} +if settings.ENVIRONMENT not in SHOW_DOCS_IN: + app_kwargs["openapi_url"] = None # disables /docs and /redoc + +app = FastAPI(**app_kwargs) +``` + +### Document endpoints fully +```python +from fastapi import APIRouter, status + +router = APIRouter() + + +@router.post( + "/items", + response_model=ItemResponse, + status_code=status.HTTP_201_CREATED, + summary="Create an item", + description="Creates an item owned by the authenticated user.", + tags=["items"], + responses={ + status.HTTP_400_BAD_REQUEST: {"model": ErrorResponse, "description": "Validation error"}, + status.HTTP_409_CONFLICT: {"model": ErrorResponse, "description": "Slug already exists"}, + }, +) +async def create_item(payload: ItemCreate) -> ItemResponse: ... +``` + ## Linting -Use ruff for formatting and linting: ```shell ruff check --fix src ruff format src ``` -## Quick Reference +Add to a pre-commit hook or run in CI. Ruff replaces black + isort + autoflake + most of flake8. -| Scenario | Solution | -|----------|----------| -| Non-blocking I/O | `async def` route with `await` | -| Blocking I/O | `def` route (sync) | -| Sync library in async | `run_in_threadpool()` | -| CPU-intensive | Celery/multiprocessing | -| Request validation | Dependencies with DB checks | -| Shared validation | Chain dependencies | -| Config per domain | Separate `BaseSettings` classes | -| Complex DB queries | SQL with JSON aggregation | +--- + +## Anti-patterns — common AI-agent mistakes + +If you're an agent reviewing a diff, check for these. Each is a real failure mode I've +seen agents introduce. + +| Anti-pattern | Why it's wrong | Fix | +|---|---|---| +| `requests.get(...)` inside `async def` | Blocks the event loop. `requests` is sync. | Use `httpx.AsyncClient` or `await run_in_threadpool(requests.get, ...)`. | +| `time.sleep` / `open()` / sync DB driver inside `async def` | Same — blocks the loop. | Use the async equivalent (`asyncio.sleep`, `aiofiles`, async driver). | +| `from jose import jwt` | `python-jose` is unmaintained. | `import jwt` (PyJWT). | +| `from async_asgi_testclient import TestClient` | Unmaintained. | `httpx.AsyncClient` + `ASGITransport`. | +| `model_config = ConfigDict(json_encoders={...})` | Deprecated in Pydantic v2. | `@field_serializer` or `Annotated[T, PlainSerializer(...)]`. | +| `Field(ge=18, default=None)` | Constraint contradicts the default. | Pick required or optional, not both. | +| `def get_user(id: int = Depends(...))` (default-arg form) | Legacy; gotchas with default values. | `user: Annotated[User, Depends(...)]`. | +| Catching `Exception` around a route's body | Hides bugs and turns 500s into silent 200s. | Catch the specific exception class; raise `HTTPException` with a meaningful status. | +| `BackgroundTasks` for anything you'd page on | No retry, dies with the worker. | Use Celery / Arq / RQ. | +| Calling a sync ORM session inside `async def` | Blocks the loop, may deadlock the pool. | Use `AsyncSession`. | +| Returning a Pydantic model and *also* setting `response_model=` to that same class | Model gets constructed twice (validate + serialize). | Either return a `dict`/ORM row and let `response_model` validate, or drop `response_model` and trust the return type. | +| Importing across domains via deep paths (`from src.auth.service.user import ...`) | Tight coupling, hard to refactor. | `from src.auth import service as auth_service`. | +| Reusing one `BaseSettings` for the whole app | Hard to reason about, every domain reads every var. | One `BaseSettings` per domain. | +| Mocking the database in integration tests | Mock/prod divergence eventually fires in prod. | Use a real DB (testcontainers, ephemeral schema) and `dependency_overrides` for auth/external services. | + +## Quick reference + +| Scenario | Solution | +|--------------------------------------|---------------------------------------------------| +| Non-blocking I/O | `async def` route with `await` | +| Blocking I/O (no async client) | `def` route (sync, runs in threadpool) | +| Sync library inside async route | `await run_in_threadpool(fn, *args)` | +| CPU-intensive work | Celery / Arq / RQ worker process | +| Request validation against DB | Dependency that loads + validates + returns | +| Reuse validation across routes | Chain dependencies | +| Inject dependency in modern style | `Annotated[T, Depends(...)]` | +| Per-request dep caching | Default behavior — same `Depends(x)` runs once | +| Per-domain config | One `BaseSettings` subclass per domain | +| Custom datetime serialization | `@field_serializer` | +| Fire-and-forget short task | `BackgroundTasks` | +| Reliable / scheduled / heavy task | Celery / Arq / RQ | +| JWT decode | `PyJWT` (`import jwt`) | +| Async DB | SQLAlchemy 2.0 async (`AsyncSession`) | +| HTTP test client | `httpx.AsyncClient` + `ASGITransport` | +| Swap dep in tests | `app.dependency_overrides[dep] = fake` | +| Lint + format | `ruff check --fix` + `ruff format` | diff --git a/README.md b/README.md index 765d78e..cd8042a 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,12 @@ After several years of building production systems, we've made both good and bad decisions that significantly impacted our developer experience. Here are some lessons worth sharing. +> **Working with an AI agent?** See [AGENTS.md](./AGENTS.md) for the same rules in a +> terse, machine-readable format with a version matrix, Do/Don't blocks, and an +> anti-patterns checklist. + +**Tested against:** Python 3.11+ · FastAPI 0.115+ · Pydantic 2.7+ · SQLAlchemy 2.0+ · Alembic 1.13+ · httpx 0.27+ · PyJWT 2.9+ · ruff 0.6+ + *[简体中文](./README_ZH.md)* ## Contents @@ -25,6 +31,7 @@ Here are some lessons worth sharing. - [Follow the REST](#follow-the-rest) - [FastAPI response serialization](#fastapi-response-serialization) - [If you must use sync SDK, then run it in a thread pool.](#if-you-must-use-sync-sdk-then-run-it-in-a-thread-pool) + - [BackgroundTasks vs a real task queue](#backgroundtasks-vs-a-real-task-queue) - [ValueErrors might become Pydantic ValidationError](#valueerrors-might-become-pydantic-validationerror) - [Docs](#docs) - [Set DB keys naming conventions](#set-db-keys-naming-conventions) @@ -211,7 +218,7 @@ class UserBase(BaseModel): first_name: str = Field(min_length=1, max_length=128) username: str = Field(min_length=1, max_length=128, pattern="^[A-Za-z0-9-_]+$") email: EmailStr - age: int = Field(ge=18, default=None) # must be greater or equal to 18 + age: int = Field(ge=18) # required, must be greater or equal to 18 favorite_band: MusicBand | None = None # only "AEROSMITH", "QUEEN", "AC/DC" values are allowed to be inputted website: AnyUrl | None = None ``` @@ -219,24 +226,23 @@ class UserBase(BaseModel): Having a controllable global base model allows us to customize all the models within the app. For instance, we can enforce a standard datetime format or introduce a common method for all subclasses of the base model. ```python from datetime import datetime +from typing import Any from zoneinfo import ZoneInfo from fastapi.encoders import jsonable_encoder -from pydantic import BaseModel, ConfigDict - - -def datetime_to_gmt_str(dt: datetime) -> str: - if not dt.tzinfo: - dt = dt.replace(tzinfo=ZoneInfo("UTC")) - - return dt.strftime("%Y-%m-%dT%H:%M:%S%z") +from pydantic import BaseModel, ConfigDict, field_serializer class CustomModel(BaseModel): - model_config = ConfigDict( - json_encoders={datetime: datetime_to_gmt_str}, - populate_by_name=True, - ) + model_config = ConfigDict(populate_by_name=True) + + @field_serializer("*", when_used="json", check_fields=False) + def _serialize_datetimes(self, value: Any) -> Any: + if isinstance(value, datetime): + if value.tzinfo is None: + value = value.replace(tzinfo=ZoneInfo("UTC")) + return value.strftime("%Y-%m-%dT%H:%M:%S%z") + return value def serializable_dict(self, **kwargs): """Return a dict which contains only serializable fields.""" @@ -273,7 +279,7 @@ auth_settings = AuthConfig() # src.config -from pydantic import PostgresDsn, RedisDsn, model_validator +from pydantic import PostgresDsn, RedisDsn from pydantic_settings import BaseSettings from src.constants import Environment @@ -345,7 +351,8 @@ Dependencies can use other dependencies and avoid code repetition for similar lo ```python # dependencies.py from fastapi.security import OAuth2PasswordBearer -from jose import JWTError, jwt +import jwt # PyJWT +from jwt.exceptions import InvalidTokenError async def valid_post_id(post_id: UUID4) -> dict[str, Any]: post = await service.get_by_id(post_id) @@ -360,7 +367,7 @@ async def parse_jwt_data( ) -> dict[str, Any]: try: payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"]) - except JWTError: + except InvalidTokenError: raise InvalidCredentials() return {"user_id": payload["id"]} @@ -397,7 +404,8 @@ but `parse_jwt_data` is called only once, in the very first call. # dependencies.py from fastapi import BackgroundTasks from fastapi.security import OAuth2PasswordBearer -from jose import JWTError, jwt +import jwt # PyJWT +from jwt.exceptions import InvalidTokenError async def valid_post_id(post_id: UUID4) -> Mapping: post = await service.get_by_id(post_id) @@ -412,7 +420,7 @@ async def parse_jwt_data( ) -> dict: try: payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"]) - except JWTError: + except InvalidTokenError: raise InvalidCredentials() return {"user_id": payload["id"]} @@ -558,6 +566,28 @@ async def call_my_sync_library(): await run_in_threadpool(client.make_request, data=my_data) ``` +### BackgroundTasks vs a real task queue +FastAPI's `BackgroundTasks` is convenient — and a footgun if you misuse it. Tasks run **after the response is sent, in the same worker process**. If the worker dies, the task is gone. There is no retry, no visibility, no scheduling. + +| Use `BackgroundTasks` when… | Use Celery / Arq / RQ when… | +|--------------------------------------------------|-------------------------------------------------| +| Task is short (< 1 second) | Task takes seconds to minutes | +| Failure can be silently dropped | You need retries or dead-letter handling | +| It's in-process (send email, log a row) | It's CPU-heavy or needs a separate worker pool | +| You don't need scheduling or rate limiting | You need cron, ETA, or rate limiting | + +```python +from fastapi import BackgroundTasks + + +@router.post("/signup") +async def signup(data: SignupIn, bg: BackgroundTasks): + user = await service.create_user(data) + bg.add_task(send_welcome_email, user.email) # fire-and-forget, in-process + return user +``` +Rule of thumb: if you'd page someone when the task is lost, it doesn't belong in `BackgroundTasks`. + ### ValueErrors might become Pydantic ValidationError If you raise a `ValueError` in a Pydantic schema that's used directly in a request body, FastAPI will return a detailed validation error response to users. ```python @@ -686,6 +716,8 @@ Being consistent with names is important. Some rules we followed: - Usually, database handles data processing much faster and cleaner than CPython will ever do. - It's preferable to do all the complex joins and simple data manipulations with SQL. - It's preferable to aggregate JSONs in DB for responses with nested objects. + +For new projects, reach for SQLAlchemy 2.0's async API (`AsyncSession`, `async_sessionmaker`). The example below uses `encode/databases` for brevity — the SQL-first principle is what matters; the client is interchangeable. ```python # src.posts.service from typing import Any @@ -769,28 +801,50 @@ async def get_creator_posts(creator: dict[str, Any] = Depends(valid_creator_id)) ``` ### Set tests client async from day 0 Writing integration tests with DB will likely lead to messed up event loop errors in the future. -Set the async test client immediately, e.g. [httpx](https://github.com/encode/starlette/issues/652) +Set the async test client immediately, using [httpx](https://www.python-httpx.org/) with `ASGITransport`. Don't reach for `async_asgi_testclient` — it's unmaintained. + ```python +from typing import AsyncGenerator + import pytest -from async_asgi_testclient import TestClient +from httpx import AsyncClient, ASGITransport from src.main import app # inited FastAPI app @pytest.fixture -async def client() -> AsyncGenerator[TestClient, None]: - host, port = "127.0.0.1", "9000" - - async with AsyncClient(transport=ASGITransport(app=app, client=(host, port)), base_url="http://test") as client: - yield client +async def client() -> AsyncGenerator[AsyncClient, None]: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as ac: + yield ac @pytest.mark.asyncio -async def test_create_post(client: TestClient): +async def test_create_post(client: AsyncClient): resp = await client.post("/posts") assert resp.status_code == 201 ``` + +#### Override dependencies in tests +Don't monkeypatch internals. FastAPI's `dependency_overrides` lets you swap any dependency for a test fake — auth, external clients, anything you don't want hitting the network. + +```python +from src.auth.dependencies import parse_jwt_data +from src.main import app + + +def fake_user(): + return {"user_id": "00000000-0000-0000-0000-000000000001"} + + +@pytest.fixture(autouse=True) +def _override_auth(): + app.dependency_overrides[parse_jwt_data] = fake_user + yield + app.dependency_overrides.clear() +``` + Unless you have synchronous database connections (excuse me?) or don't plan to write integration tests. ### Use ruff diff --git a/README_ZH.md b/README_ZH.md index c224939..59309f3 100755 --- a/README_ZH.md +++ b/README_ZH.md @@ -222,7 +222,7 @@ class UserBase(BaseModel): first_name: str = Field(min_length=1, max_length=128) username: str = Field(min_length=1, max_length=128, pattern="^[A-Za-z0-9-_]+$") email: EmailStr - age: int = Field(ge=18, default=None) # 必须大于或等于18 + age: int = Field(ge=18) # 必须大于或等于18 favorite_band: MusicBand | None = None # 只允许输入"AEROSMITH"、"QUEEN"、"AC/DC"值 website: AnyUrl | None = None ``` @@ -233,22 +233,22 @@ class UserBase(BaseModel): ```python from datetime import datetime +from typing import Any from zoneinfo import ZoneInfo from fastapi.encoders import jsonable_encoder -from pydantic import BaseModel, ConfigDict - -def datetime_to_gmt_str(dt: datetime) -> str: - if not dt.tzinfo: - dt = dt.replace(tzinfo=ZoneInfo("UTC")) - - return dt.strftime("%Y-%m-%dT%H:%M:%S%z") +from pydantic import BaseModel, ConfigDict, field_serializer class CustomModel(BaseModel): - model_config = ConfigDict( - json_encoders={datetime: datetime_to_gmt_str}, - populate_by_name=True, - ) + model_config = ConfigDict(populate_by_name=True) + + @field_serializer("*", when_used="json", check_fields=False) + def _serialize_datetimes(self, value: Any) -> Any: + if isinstance(value, datetime): + if value.tzinfo is None: + value = value.replace(tzinfo=ZoneInfo("UTC")) + return value.strftime("%Y-%m-%dT%H:%M:%S%z") + return value def serializable_dict(self, **kwargs): """返回仅包含可序列化字段的字典。""" @@ -285,7 +285,7 @@ class AuthConfig(BaseSettings): auth_settings = AuthConfig() # src.config -from pydantic import PostgresDsn, RedisDsn, model_validator +from pydantic import PostgresDsn, RedisDsn from pydantic_settings import BaseSettings from src.constants import Environment @@ -356,7 +356,8 @@ async def get_post_reviews(post: dict[str, Any] = Depends(valid_post_id)): ```python # dependencies.py from fastapi.security import OAuth2PasswordBearer -from jose import JWTError, jwt +import jwt # PyJWT +from jwt.exceptions import InvalidTokenError async def valid_post_id(post_id: UUID4) -> dict[str, Any]: post = await service.get_by_id(post_id) @@ -370,7 +371,7 @@ async def parse_jwt_data( ) -> dict[str, Any]: try: payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"]) - except JWTError: + except InvalidTokenError: raise InvalidCredentials() return {"user_id": payload["id"]} @@ -408,7 +409,8 @@ async def get_user_post(post: dict[str, Any] = Depends(valid_owned_post)): # dependencies.py from fastapi import BackgroundTasks from fastapi.security import OAuth2PasswordBearer -from jose import JWTError, jwt +import jwt # PyJWT +from jwt.exceptions import InvalidTokenError async def valid_post_id(post_id: UUID4) -> Mapping: post = await service.get_by_id(post_id) @@ -422,7 +424,7 @@ async def parse_jwt_data( ) -> dict: try: payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"]) - except JWTError: + except InvalidTokenError: raise InvalidCredentials() return {"user_id": payload["id"]} @@ -804,20 +806,21 @@ async def get_creator_posts(creator: dict[str, Any] = Depends(valid_creator_id)) 使用数据库编写集成测试很可能在将来导致混乱的事件循环错误。立即设置异步测试客户端，例如[httpx](https://github.com/encode/starlette/issues/652) ```python +from typing import AsyncGenerator + import pytest -from async_asgi_testclient import TestClient +from httpx import AsyncClient, ASGITransport from src.main import app # inited FastAPI app @pytest.fixture -async def client() -> AsyncGenerator[TestClient, None]: - host, port = "127.0.0.1", "9000" - - async with AsyncClient(transport=ASGITransport(app=app, client=(host, port)), base_url="http://test") as client: - yield client +async def client() -> AsyncGenerator[AsyncClient, None]: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as ac: + yield ac @pytest.mark.asyncio -async def test_create_post(client: TestClient): +async def test_create_post(client: AsyncClient): resp = await client.post("/posts") assert resp.status_code == 201