Quentin/doluedu-fastapi-template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 改造为开箱即用模板 — MySQL/MongoDB(Beanie) 一键切换 + AGENTS.md 开发规范

Browse Source 
- src/ 按域组织骨架，示例域 items 双后端实现（SQLAlchemy2.0 async / Beanie 2.x）
- DB_BACKEND 环境变量切换，router 只依赖 Repo Protocol
- alembic 迁移（含初始 items 迁移）、uv 依赖管理、ruff、pytest 异步测试
- Dockerfile + docker-compose（app/mysql/mongo）
- AGENTS.md 重写为本模板开发规范；原最佳实践文档归档 docs/
- 验证：mongodb/mysql 双后端 pytest 全绿 + uvicorn 真实冒烟通过
This commit is contained in:
ghplvh
2026-08-21 21:23:26 +08:00
parent 5e00aa6095
commit 981bee5a46
34 changed files with 3579 additions and 1256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
.env.example Normal file
View File

@@ -0,0 +1,19 @@
# ===== 数据库后端：mysql 或 mongodb =====
DB_BACKEND=mongodb
# ===== MySQL（DB_BACKEND=mysql 时必填）=====
MYSQL_DSN=mysql+aiomysql://root:root@127.0.0.1:3306/app?charset=utf8mb4
# ===== MongoDB（DB_BACKEND=mongodb 时必填）=====
MONGO_DSN=mongodb://127.0.0.1:27017
MONGO_DB=app
# ===== 应用 =====
APP_NAME=fastapi-template
APP_ENV=dev
APP_DEBUG=true
APP_HOST=0.0.0.0
APP_PORT=8000
# 跨域，逗号分隔；* 仅调试用
CORS_ORIGINS=*

9
.gitignore vendored Normal file
View File

@@ -0,0 +1,9 @@
__pycache__/
*.pyc
.venv/
.env
.DS_Store
.ruff_cache/
.pytest_cache/
dist/
*.egg-info/

513
AGENTS.md
View File

@@ -1,447 +1,144 @@
# FastAPI Best Practices for AI Agents # AGENTS.md — fastapi-template 开发规范
A machine-readable companion to [README.md](./README.md) for AI coding agents 本文件是 AI Agent / 开发者在本仓库工作的**必读规范**。先读这个，再动手。
working in FastAPI projects. Same rules, restructured for fast pattern matching: 通用 FastAPI 最佳实践参考 [docs/AGENTS_GENERIC.md](./docs/AGENTS_GENERIC.md) 与 [docs/BEST_PRACTICES_ZH.md](./docs/BEST_PRACTICES_ZH.md)。
version pins, Do/Don't blocks, anti-patterns, and a quick-reference table.
## Compatibility Matrix ## 0. 这个模板是什么
Pin to these versions or newer. Examples in this file assume them. FastAPI 生产级项目骨架。**核心特性：`DB_BACKEND` 一个环境变量切换 MySQL / MongoDB，业务代码零改动**。
任何修改都不得破坏这个特性——router/schemas 永远不允许 import 具体 DB 实现。
| Dependency | Minimum | Notes | ## 1. 技术栈基线
|------------------|-----------|------------------------------------------------------|
| 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 | 项 | 版本 | 说明 |
|---|---|---|
| Python | ≥3.11 | 用 `X \| Y`、`StrEnum` |
| FastAPI | ≥0.115 | 依赖注入一律 `Annotated[T, Depends(...)]` |
| Pydantic | v2 | 禁用 v1 API（`.dict()` / `json_encoders`） |
| SQLAlchemy | 2.0 async | `AsyncSession` + `async_sessionmaker`，禁 sync session |
| Beanie | ≥1.27 | MongoDB ODM |
| 包管理 | **uv** | 禁 pip/poetry/conda；加依赖 `uv add xxx` |
| Lint | ruff | 提交前 `uv run ruff check --fix .` 必须零告警 |
| 测试 | pytest + pytest-asyncio + httpx | `asyncio_mode=auto` |
Organize by domain, not by file type. One package per bounded context. ## 2. 目录结构铁律
按**域（domain）**组织，不按文件类型。一个域一个包，照抄 `src/items/`：
``` ```
src/ src/{domain}/
├── {domain}/ # e.g., auth/, posts/, aws/ ├── router.py # 路由：只做参数校验和调用 repo，禁写 SQL/查询
│ ├── router.py # API endpoints ├── schemas.py # Pydantic 契约：与 DB 无关，前后端共用语义
│ ├── schemas.py # Pydantic models ├── dependencies.py # 定义 Repo Protocol + 按 DB_BACKEND 选实现
│ ├── models.py # SQLAlchemy ORM models ├── mysql.py # MySQL ORM 模型 + MySQLXxxRepo
│ ├── service.py # Business logic ├── mongo.py # Beanie Document + MongoXxxRepo
│ ├── dependencies.py # Route dependencies ├── service.py # （可选）跨 repo 的复杂业务逻辑
│ ├── config.py # Domain-scoped BaseSettings ├── constants.py # （可选）常量、错误码
│ ├── constants.py # Constants and error codes └── exceptions.py # （可选）域内异常，继承 src.exceptions.AppError
│ ├── exceptions.py # Domain-specific exceptions
│ └── utils.py # Helper functions
├── config.py # Global BaseSettings
├── models.py # Shared Pydantic / ORM bases
├── exceptions.py # Global exceptions
├── database.py # Async engine + session factory
└── main.py # FastAPI app + lifespan
``` ```
**Cross-domain imports**: always use the explicit module name. Never `from src.auth import *`. - 新增域后必做三件事：① `src/mongo.py::_collect_documents()` 登记 Document；
② `alembic/env.py` import mysql 模型；③ `src/main.py` include_router。
- 跨域 import 必须显式模块名：`from src.items import constants as item_constants`，禁 `import *`。
- 全局共享的东西才放 `src/` 根（config / exceptions / database / mongo）。
## 3. 双数据库后端规范（本模板灵魂）
```python ```python
from src.auth import constants as auth_constants # ✅ router 只依赖 Protocol
from src.notifications import service as notification_service async def get_item(item_id: str, repo: ItemRepoDep): ...
from src.posts.constants import ErrorCode as PostsErrorCode
# ✅ dependencies.py 模块加载时选定实现
get_item_repo = _get_mysql_repo if settings.DB_BACKEND == "mysql" else _get_mongo_repo
# ❌ 禁止在 router/schemas/service 里出现
from src.items.mysql import Item # 不许！
from src.items.mongo import ItemDoc # 不许！
``` ```
## Async Routes - 两个 repo 实现**同一 Protocol**，方法签名逐字一致。
- schemas 的 `id` 统一用 `str`（MySQL int 自增 / Mongo ObjectId 都转 str），
转换在各 repo 的 `to_out()` 里完成。
- MySQL 改表结构 → 必须 `alembic revision --autogenerate` 生成迁移，禁手改生产库。
- Mongo 新增 Document → 只登记 `_collect_documents()`，beanie 自动建索引。
- `src/database.py`、`src/mongo.py` 只在 lifespan / dependencies 里被引用；
非对应后端模式下它们可以 import 但**不得产生连接**（engine 惰性，别主动 connect）。
### Decision rule ## 4. 异步纪律
| Route does this | Use | | 场景 | 写法 |
|----------------------------------------|-------------| |---|---|
| `await`-able non-blocking I/O | `async def` | | 可 await 的 I/O | `async def` |
| Blocking I/O (no async client exists) | `def` (sync, runs in threadpool) | | 只有同步 SDK | `def`（FastAPI 自动进线程池）或 `run_in_threadpool` |
| Mix of both | `async def` + `run_in_threadpool` for the blocking part | | CPU 密集 >50ms | 扔任务队列，别放请求路径 |
| CPU-bound work (>50 ms compute) | Offload to a worker process (Celery / RQ / Arq) |
### Do / Don't
```python ```python
# DON'T — blocking call inside async route freezes the entire event loop # ❌ async 路由里调用阻塞库 = 冻结整个事件循环
@router.get("/bad") @router.get("/bad")
async def bad(): async def bad():
time.sleep(10) # blocks every request on this worker time.sleep(5)
return {"ok": True}
# DO — sync route lets FastAPI run it in a threadpool # ✅
@router.get("/sync-ok") @router.get("/ok")
def sync_ok(): def ok():
time.sleep(10) # blocks one threadpool worker, not the loop time.sleep(5)
return {"ok": True}
# 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}
# DO — async route that has to call a sync library
from fastapi.concurrency import run_in_threadpool
@router.get("/wrap")
async def wrap():
result = await run_in_threadpool(legacy_sync_client.fetch, "id")
return result
``` ```
### Threadpool caveats ## 5. 配置与异常
- 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 - 配置只走 `src/config.py` 的 `settings`（pydantic-settings）；域级配置放 `{domain}/config.py`，`env_prefix="XXX_"`。
- 禁在代码里硬编码 DSN / 密钥；`.env` 不入库，`.env.example` 必须同步更新。
- 业务异常继承 `AppError`，404 用 `NotFoundError`；统一由 `register_exception_handlers` 输出 `{"detail": ...}`。
- 禁裸 `except:`，禁 `except Exception: pass`。
### Use built-in validators ## 6. Pydantic 规范
```python
from enum import StrEnum
from pydantic import AnyUrl, BaseModel, EmailStr, Field
class MusicBand(StrEnum):
AEROSMITH = "AEROSMITH"
QUEEN = "QUEEN"
ACDC = "AC/DC"
class UserCreate(BaseModel):
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) # required, must be >= 18
favorite_band: MusicBand | None = None
website: AnyUrl | None = None
```
> **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 ```python
from datetime import datetime # ✅ 约束写清楚；required 和 optional 二选一，别自相矛盾
from zoneinfo import ZoneInfo name: str = Field(min_length=1, max_length=128)
from pydantic import BaseModel, ConfigDict, field_serializer age: int | None = Field(default=None, ge=0) # 可选
age: int = Field(ge=18) # 必填
# ❌ 矛盾写法
class CustomModel(BaseModel): age: int = Field(ge=18, default=None)
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 - 序列化定制用 `@field_serializer`，不用 `json_encoders`。
- 入参模型（XxxCreate/XxxUpdate）与出参模型（XxxOut）分开；`XxxUpdate` 全字段可选 + `model_dump(exclude_unset=True)`。
`pydantic-settings` is its own package since Pydantic v2. ## 7. 测试规范
```python - 测试从第一天就是异步：`httpx.AsyncClient` + `ASGITransport`，见 `tests/conftest.py`。
# src/auth/config.py - 测试库与开发库隔离：Mongo 用 `app_test`，MySQL 用 `app_test`。
from datetime import timedelta - 每个域至少覆盖：create / get / list / update / delete + 404 路径。
from pydantic_settings import BaseSettings, SettingsConfigDict - 跑 MySQL 后端测试：`DB_BACKEND=mysql uv run pytest`。
## 8. Git 规范
class AuthConfig(BaseSettings): - 提交信息：Conventional Commits —— `feat: xxx` / `fix: xxx` / `refactor:` / `docs:` / `test:` / `chore:`。
model_config = SettingsConfigDict(env_prefix="AUTH_", env_file=".env", extra="ignore") - 主分支 `main`；功能开发开 `feature/xxx` 分支，提 PR 合并。
- 提交前自检三连：`uv run ruff check --fix . && uv run pytest`，全绿才推。
JWT_ALG: str ## 9. 常用命令速查
JWT_SECRET: str
JWT_EXP_MINUTES: int = 5
REFRESH_TOKEN_KEY: str
REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)
SECURE_COOKIES: bool = True
```bash
auth_settings = AuthConfig() uv sync # 装依赖
uv add fastapi # 加依赖（dev 依赖：uv add --dev xxx）
uv run uvicorn src.main:app --reload # 开发起服务
uv run pytest # 测试
uv run ruff check --fix . # lint
uv run alembic revision --autogenerate -m "msg" # MySQL 迁移
docker compose up -d --build # 容器化整套
``` ```
## Dependencies ## 10. 反模式清单（Review 时逐条核对）
### Use Annotated, not default-arg `Depends(...)` - [ ] router 里出现 SQL / `ItemDoc.find()` 等具体 DB 调用
- [ ] async 路由里调用 requests / time.sleep 等阻塞库
`Annotated[T, Depends(...)]` is the idiomatic form since FastAPI 0.95 and avoids - [ ] `print` 调试残留（用 `logging`）
gotchas with default values. - [ ] 硬编码连接串 / 密钥
- [ ] `from x import *`、裸 except
```python - [ ] 改了 MySQL 模型没生成 alembic 迁移
# DO — modern Annotated form - [ ] 新增 Mongo Document 没登记 `_collect_documents()`
from typing import Annotated - [ ] 新域只在单一后端实现（两个 repo 必须成对）
from fastapi import Depends - [ ] 提交信息 freestyle（非 Conventional Commits）
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
```
### Chain dependencies for reuse
```python
async def valid_owned_post(
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
```
### 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}`).
## Authentication — JWT
Use **`PyJWT`**, not `python-jose` (unmaintained).
```python
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 — SQLAlchemy 2.0 async
Prefer SQLAlchemy 2.0's async API. `encode/databases` is in maintenance mode — don't pick it for new projects.
```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",
"ck": "%(table_name)s_%(constraint_name)s_check",
"fk": "%(table_name)s_%(column_0_name)s_fkey",
"pk": "%(table_name)s_pkey",
}
metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)
```
### 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.
## Background work — BackgroundTasks vs Celery
| 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 |
```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
```
> 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
### Async client from day one
```python
import pytest
from httpx import AsyncClient, ASGITransport
from src.main import app
@pytest.fixture
async def client():
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: 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
```shell
ruff check --fix src
ruff format src
```
Add to a pre-commit hook or run in CI. Ruff replaces black + isort + autoflake + most of flake8.
---
## 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` |

16
Dockerfile Normal file
View File

@@ -0,0 +1,16 @@
FROM python:3.12-slim
WORKDIR /app
# uv 管理依赖（国内构建加 --index-url 镜像）
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
COPY pyproject.toml uv.lock* ./
RUN uv sync --no-dev --no-cache
COPY src ./src
COPY alembic ./alembic
COPY alembic.ini ./
EXPOSE 8000
CMD ["uv", "run", "--no-sync", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

912
README.md
View File

@@ -1,871 +1,87 @@
## FastAPI Best Practices <!-- omit from toc --> # fastapi-template
Opinionated list of best practices and conventions we use at our startups.
After several years of building production systems, 开箱即用的 FastAPI 项目模板。**一条环境变量切换 MySQL / MongoDB**，业务代码零改动。
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 - MySQL：SQLAlchemy 2.0 async + aiomysql + Alembic 迁移
> terse, machine-readable format with a version matrix, Do/Don't blocks, and an - MongoDB：Beanie ODM + Motor
> anti-patterns checklist. - 工程化：uv 管理依赖、ruff lint、pytest 异步测试、Docker / compose 一键起
*[简体中文](./README_ZH.md)* > 写给 AI Agent 的开发规范见 [AGENTS.md](./AGENTS.md)。
> FastAPI 通用最佳实践（原版文档）：[docs/BEST_PRACTICES_ZH.md](./docs/BEST_PRACTICES_ZH.md)
## Contents <!-- omit from toc --> ## 快速开始
- [Project Structure](#project-structure)
- [Async Routes](#async-routes)
- [I/O Intensive Tasks](#io-intensive-tasks)
- [CPU Intensive Tasks](#cpu-intensive-tasks)
- [Pydantic](#pydantic)
- [Excessively use Pydantic](#excessively-use-pydantic)
- [Custom Base Model](#custom-base-model)
- [Decouple Pydantic BaseSettings](#decouple-pydantic-basesettings)
- [Dependencies](#dependencies)
- [Beyond Dependency Injection](#beyond-dependency-injection)
- [Chain Dependencies](#chain-dependencies)
- [Decouple \& Reuse dependencies. Dependency calls are cached](#decouple--reuse-dependencies-dependency-calls-are-cached)
- [Prefer `async` dependencies](#prefer-async-dependencies)
- [Miscellaneous](#miscellaneous)
- [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)
- [Migrations. Alembic](#migrations-alembic)
- [Set DB naming conventions](#set-db-naming-conventions)
- [SQL-first. Pydantic-second](#sql-first-pydantic-second)
- [Set tests client async from day 0](#set-tests-client-async-from-day-0)
- [Use ruff](#use-ruff)
- [Bonus Section](#bonus-section)
## Project Structure ```bash
There are many ways to structure a project, but the best structure is one that is consistent, straightforward, and free of surprises. # 1. 克隆后改个名
git clone https://git.code-lab.cn/Quentin/fastapi-template.git my-project && cd my-project
Many example projects and tutorials organize projects by file type (e.g., crud, routers, models), which works well for microservices or smaller projects. However, this approach didn't scale well for our monolith with many domains and modules. # 2. 配置：选数据库后端
cp .env.example .env
# 编辑 .env：DB_BACKEND=mongodb 或 mysql，填对应 DSN
The structure I found more scalable and evolvable is inspired by Netflix's [Dispatch](https://github.com/Netflix/dispatch), with some minor modifications. # 3. 装依赖（uv）
``` uv sync
fastapi-project
├── alembic/ # 4. 起数据库（或直接用现成的）
├── src docker compose up -d mongo # MongoDB
│ ├── auth docker compose up -d mysql # MySQL
│ │ ├── router.py
│ │ ├── schemas.py # pydantic models # 5. 跑！
│ │ ├── models.py # db models uv run uvicorn src.main:app --reload
│ │ ├── dependencies.py
│ │ ├── config.py # local configs
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ ├── service.py
│ │ └── utils.py
│ ├── aws
│ │ ├── client.py # client model for external service communication
│ │ ├── schemas.py
│ │ ├── config.py
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ └── utils.py
│ ├── posts
│ │ ├── router.py
│ │ ├── schemas.py
│ │ ├── models.py
│ │ ├── dependencies.py
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ ├── service.py
│ │ └── utils.py
│ ├── config.py # global configs
│ ├── models.py # global models
│ ├── exceptions.py # global exceptions
│ ├── pagination.py # global module e.g. pagination
│ ├── database.py # db connection related stuff
│ └── main.py
├── tests/
│ ├── auth
│ ├── aws
│ └── posts
├── templates/
│ └── index.html
├── requirements
│ ├── base.txt
│ ├── dev.txt
│ └── prod.txt
├── .env
├── .gitignore
├── logging.ini
└── alembic.ini
```
1. Store all domain directories inside `src` folder
1. `src/` - highest level of an app, contains common models, configs, and constants, etc.
2. `src/main.py` - root of the project, which inits the FastAPI app
2. Each package has its own router, schemas, models, etc.
1. `router.py` - is a core of each module with all the endpoints
2. `schemas.py` - for pydantic models
3. `models.py` - for db models
4. `service.py` - module specific business logic
5. `dependencies.py` - router dependencies
6. `constants.py` - module specific constants and error codes
7. `config.py` - e.g. env vars
8. `utils.py` - non-business logic functions, e.g. response normalization, data enrichment, etc.
9. `exceptions.py` - module specific exceptions, e.g. `PostNotFound`, `InvalidUserData`
3. When package requires services or dependencies or constants from other packages - import them with an explicit module name
```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 # in case we have Standard ErrorCode in constants module of each package
``` ```
## Async Routes 打开 http://127.0.0.1:8000/docs 看交互式 API 文档，
FastAPI is an async-first framework—it's designed to work with async I/O operations, which is why it's so fast. `GET /health` 会返回当前生效的 `db_backend`。
However, FastAPI doesn't restrict you to only `async` routes; you can use `sync` routes too. This might confuse beginners into thinking they're the same, but they're not. ## 切库说明
### I/O Intensive Tasks | | MySQL | MongoDB |
Under the hood, FastAPI can [effectively handle](https://fastapi.tiangolo.com/async/#path-operation-functions) both async and sync I/O operations: |---|---|---|
- FastAPI runs `sync` routes in a [threadpool](https://en.wikipedia.org/wiki/Thread_pool), so blocking I/O operations won't stop the [event loop](https://docs.python.org/3/library/asyncio-eventloop.html) from executing other tasks. | 开关 | `DB_BACKEND=mysql` | `DB_BACKEND=mongodb` |
- If the route is defined as `async`, it's called via `await` and FastAPI trusts you to only perform non-blocking I/O operations. | 连接 | `MYSQL_DSN=mysql+aiomysql://user:pass@host:3306/db` | `MONGO_DSN` + `MONGO_DB` |
| 模型 | `src/{domain}/mysql.py`（ORM） | `src/{domain}/mongo.py`（Document） |
| 迁移 | `alembic revision --autogenerate` + `upgrade head` | 不需要（beanie 自动建索引） |
The caveat is that if you violate that trust and execute blocking operations within async routes, the event loop won't be able to run other tasks until the blocking operation completes. router/service 只依赖 `ItemRepo` 协议（见 `src/items/dependencies.py`），
```python 两个后端实现同一套接口，`.env` 改一行即切换。
import asyncio
import time
from fastapi import APIRouter ## 项目结构
router = APIRouter()
@router.get("/terrible-ping")
async def terrible_ping():
time.sleep(10) # I/O blocking operation for 10 seconds, the whole process will be blocked
return {"pong": True}
@router.get("/good-ping")
def good_ping():
time.sleep(10) # I/O blocking operation for 10 seconds, but in a separate thread for the whole `good_ping` route
return {"pong": True}
@router.get("/perfect-ping")
async def perfect_ping():
await asyncio.sleep(10) # non-blocking I/O operation
return {"pong": True}
``` ```
**What happens when we call:** ├── src/
1. `GET /terrible-ping` │ ├── main.py # app 工厂 + lifespan（按后端初始化 DB）
1. FastAPI server receives a request and starts handling it │ ├── config.py # pydantic-settings 全局配置
2. Server's event loop and all queued tasks wait until `time.sleep()` finishes │ ├── database.py # MySQL engine/session（仅 mysql 模式使用）
1. Since the route is `async`, the server doesn't offload it to a threadpool—it blocks the entire event loop │ ├── mongo.py # beanie 初始化（仅 mongodb 模式使用）
2. Server won't accept any new requests while waiting │ ├── exceptions.py # 全局异常 + 统一错误响应
3. Server returns the response │ └── items/ # 示例域（新增域照抄这个目录）
1. Only after responding does the server resume accepting new requests │ ├── router.py # 路由：只依赖 ItemRepo 协议
2. `GET /good-ping` │ ├── schemas.py # Pydantic 契约（与 DB 无关）
1. FastAPI server receives a request and starts handling it │ ├── dependencies.py# 按 DB_BACKEND 选仓储实现
2. FastAPI sends the entire `good_ping` route to the threadpool, where a worker thread runs the function │ ├── mysql.py # MySQL 模型 + 仓储
3. While `good_ping` executes, the event loop continues processing other tasks (e.g., accepting new requests, calling the database) │ └── mongo.py # MongoDB Document + 仓储
- The worker thread waits for `time.sleep` to finish, independently of the main thread ├── alembic/ # MySQL 迁移
- The sync operation blocks only the worker thread, not the main event loop ├── tests/ # pytest + httpx ASGITransport
4. When `good_ping` finishes, the server returns a response to the client ├── Dockerfile
3. `GET /perfect-ping` └── docker-compose.yml # app + mysql + mongo
1. FastAPI server receives a request and starts handling it
2. FastAPI awaits `asyncio.sleep(10)`
3. Event loop continues processing other tasks from the queue (e.g., accepting new requests, calling the database)
4. When `asyncio.sleep(10)` completes, the server finishes executing the route and returns a response to the client
> [!WARNING]
> Notes on the thread pool:
> - Threads require more resources than coroutines, so they are not as cheap as async I/O operations.
> - Thread pool has a limited number of threads, i.e. you might run out of threads and your app will become slow. [Read more](https://github.com/Kludex/fastapi-tips?tab=readme-ov-file#2-be-careful-with-non-async-functions) (external link)
### CPU Intensive Tasks
The second caveat is that non-blocking awaitables and threadpool offloading are only beneficial for I/O intensive tasks (e.g., file operations, database calls, external API requests).
- Awaiting CPU-intensive tasks (e.g., heavy calculations, data processing, video transcoding) provides no benefit since the CPU must actively work to complete them. In contrast, I/O operations are external—the server just waits for a response and can handle other tasks in the meantime.
- Running CPU-intensive tasks in other threads is also ineffective due to the [GIL](https://realpython.com/python-gil/). In short, the GIL allows only one thread to execute Python bytecode at a time, making threads ineffective for CPU-bound work.
- To optimize CPU-intensive tasks, you should offload them to worker processes (e.g., using `multiprocessing` or a task queue like Celery).
**Related StackOverflow questions of confused users**
1. https://stackoverflow.com/questions/62976648/architecture-flask-vs-fastapi/70309597#70309597
- Here you can also check [my answer](https://stackoverflow.com/a/70309597/6927498)
2. https://stackoverflow.com/questions/65342833/fastapi-uploadfile-is-slow-compared-to-flask
3. https://stackoverflow.com/questions/71516140/fastapi-runs-api-calls-in-serial-instead-of-parallel-fashion
## Pydantic
### Excessively use Pydantic
Pydantic has a rich set of features to validate and transform data.
In addition to standard features like required and optional fields with default values,
Pydantic has built-in data processing tools like regex validation, enums, string manipulation, email validation, and more.
```python
from enum import StrEnum
from pydantic import AnyUrl, BaseModel, EmailStr, Field
class MusicBand(StrEnum):
AEROSMITH = "AEROSMITH"
QUEEN = "QUEEN"
ACDC = "AC/DC"
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) # 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
```
### Custom Base Model
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, field_serializer
class CustomModel(BaseModel):
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."""
default_dict = self.model_dump()
return jsonable_encoder(default_dict)
```
In the example above, we have decided to create a global base model that:
- Serializes all datetime fields to a standard format with an explicit timezone
- Provides a method to return a dict with only serializable fields
### Decouple Pydantic BaseSettings
BaseSettings is great for reading environment variables, but a single BaseSettings for the whole app gets messy. Split it across modules and domains.
```python
# src.auth.config
from datetime import timedelta
from pydantic_settings import BaseSettings
class AuthConfig(BaseSettings):
JWT_ALG: str
JWT_SECRET: str
JWT_EXP: int = 5 # minutes
REFRESH_TOKEN_KEY: str
REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)
SECURE_COOKIES: bool = True
auth_settings = AuthConfig()
# src.config
from pydantic import PostgresDsn, RedisDsn
from pydantic_settings import BaseSettings
from src.constants import Environment
class Config(BaseSettings):
DATABASE_URL: PostgresDsn
REDIS_URL: RedisDsn
SITE_DOMAIN: str = "myapp.com"
ENVIRONMENT: Environment = Environment.PRODUCTION
SENTRY_DSN: str | None = None
CORS_ORIGINS: list[str]
CORS_ORIGINS_REGEX: str | None = None
CORS_HEADERS: list[str]
APP_VERSION: str = "1.0"
settings = Config()
``` ```
## Dependencies ## 常用命令
### Beyond Dependency Injection
Pydantic is a great schema validator, but for complex validations that require database or external service calls, it's not enough.
FastAPI docs mostly present dependencies as DI for endpoints, but they're also great for request validation.
Dependencies can validate data against database constraints (e.g., checking if an email already exists, ensuring a user exists, etc.).
```python
# dependencies.py
async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
post = await service.get_by_id(post_id)
if not post:
raise PostNotFound()
return post
# router.py
@router.get("/posts/{post_id}", response_model=PostResponse)
async def get_post_by_id(post: dict[str, Any] = Depends(valid_post_id)):
return post
@router.put("/posts/{post_id}", response_model=PostResponse)
async def update_post(
update_data: PostUpdate,
post: dict[str, Any] = Depends(valid_post_id),
):
updated_post = await service.update(id=post["id"], data=update_data)
return updated_post
@router.get("/posts/{post_id}/reviews", response_model=list[ReviewsResponse])
async def get_post_reviews(post: dict[str, Any] = Depends(valid_post_id)):
post_reviews = await reviews_service.get_by_post_id(post["id"])
return post_reviews
```
If we didn't put data validation in a dependency, we would have to validate that `post_id` exists
in every endpoint and write the same tests for each of them.
### Chain Dependencies
Dependencies can use other dependencies and avoid code repetition for similar logic.
```python
# dependencies.py
from fastapi.security import OAuth2PasswordBearer
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)
if not post:
raise PostNotFound()
return post
async def parse_jwt_data(
token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict[str, Any]:
try:
payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
except InvalidTokenError:
raise InvalidCredentials()
return {"user_id": payload["id"]}
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]:
if post["creator_id"] != token_data["user_id"]:
raise UserNotOwner()
return post
# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(post: dict[str, Any] = Depends(valid_owned_post)):
return post
```
### Decouple & Reuse dependencies. Dependency calls are cached
Dependencies can be reused multiple times, and they won't be recalculated - FastAPI caches dependency's result within a request's scope by default,
i.e. if `valid_post_id` gets called multiple times in one route, it will be called only once.
Knowing this, we can decouple dependencies onto multiple smaller functions that operate on a smaller domain and are easier to reuse in other routes.
For example, in the code below we are using `parse_jwt_data` three times:
1. `valid_owned_post`
2. `valid_active_creator`
3. `get_user_post`,
but `parse_jwt_data` is called only once, in the very first call.
```python
# dependencies.py
from fastapi import BackgroundTasks
from fastapi.security import OAuth2PasswordBearer
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)
if not post:
raise PostNotFound()
return post
async def parse_jwt_data(
token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict:
try:
payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
except InvalidTokenError:
raise InvalidCredentials()
return {"user_id": payload["id"]}
async def valid_owned_post(
post: Mapping = Depends(valid_post_id),
token_data: dict = Depends(parse_jwt_data),
) -> Mapping:
if post["creator_id"] != token_data["user_id"]:
raise UserNotOwner()
return post
async def valid_active_creator(
token_data: dict = Depends(parse_jwt_data),
):
user = await users_service.get_by_id(token_data["user_id"])
if not user["is_active"]:
raise UserIsBanned()
if not user["is_creator"]:
raise UserNotCreator()
return user
# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(
worker: BackgroundTasks,
post: Mapping = Depends(valid_owned_post),
user: Mapping = Depends(valid_active_creator),
):
"""Get post that belong the active user."""
worker.add_task(notifications_service.send_email, user["id"])
return post
```bash
uv run pytest # 测试（默认 mongodb 后端，库名 app_test）
uv run ruff check --fix . # lint + 自动修
uv run alembic revision --autogenerate -m "msg" # 生成 MySQL 迁移
uv run alembic upgrade head # 执行迁移
docker compose up -d --build # 整套起
``` ```
### Prefer `async` dependencies ## 新增一个业务域
FastAPI supports both `sync` and `async` dependencies. It's tempting to use `sync` when you don't need to await anything, but that's not the best choice.
Just like routes, `sync` dependencies run in a threadpool. Threads have overhead that's unnecessary for small non-I/O operations. 照抄 `src/items/` 为 `src/{domain}/`，然后：
[See more](https://github.com/Kludex/fastapi-tips?tab=readme-ov-file#9-your-dependencies-may-be-running-on-threads) (external link) 1. `mongo.py` 里写 Document，并在 `src/mongo.py` 的 `_collect_documents()` 登记
2. `mysql.py` 里写 ORM 模型，并在 `alembic/env.py` import 保证 metadata 可见
3. `main.py` 里 `include_router`
详细规范（结构、命名、依赖注入、异步纪律、Git 提交）都在 [AGENTS.md](./AGENTS.md)。
## Miscellaneous
### Follow the REST
Developing RESTful API makes it easier to reuse dependencies in routes like these:
1. `GET /courses/:course_id`
2. `GET /courses/:course_id/chapters/:chapter_id/lessons`
3. `GET /chapters/:chapter_id`
The only caveat is having to use the same variable names in the path:
- If you have two endpoints `GET /profiles/:profile_id` and `GET /creators/:creator_id`
that both validate whether the given `profile_id` exists, but `GET /creators/:creator_id`
also checks if the profile is creator, then it's better to rename `creator_id` path variable to `profile_id` and chain those two dependencies.
```python
# src.profiles.dependencies
async def valid_profile_id(profile_id: UUID4) -> Mapping:
profile = await service.get_by_id(profile_id)
if not profile:
raise ProfileNotFound()
return profile
# src.creators.dependencies
async def valid_creator_id(profile: Mapping = Depends(valid_profile_id)) -> Mapping:
if not profile["is_creator"]:
raise ProfileNotCreator()
return profile
# src.profiles.router.py
@router.get("/profiles/{profile_id}", response_model=ProfileResponse)
async def get_user_profile_by_id(profile: Mapping = Depends(valid_profile_id)):
"""Get profile by id."""
return profile
# src.creators.router.py
@router.get("/creators/{profile_id}", response_model=ProfileResponse)
async def get_user_profile_by_id(
creator_profile: Mapping = Depends(valid_creator_id)
):
"""Get creator's profile by id."""
return creator_profile
```
### FastAPI response serialization
You might think you can return a Pydantic object that matches your route's `response_model` and skip some processing steps, but you'd be wrong.
FastAPI first converts the Pydantic object to a dict using `jsonable_encoder`, then validates the data against your `response_model`, and only then serializes it to JSON.
This means your Pydantic model object is created twice:
- First, when you explicitly create it to return from your route.
- Second, implicitly by FastAPI to validate the response data according to the response_model.
```python
from fastapi import FastAPI
from pydantic import BaseModel, model_validator
app = FastAPI()
class ProfileResponse(BaseModel):
@model_validator(mode="after")
def debug_usage(self):
print("created pydantic model")
return self
@app.get("/", response_model=ProfileResponse)
async def root():
return ProfileResponse()
```
**Logs Output:**
```
[INFO] [2022-08-28 12:00:00.000000] created pydantic model
[INFO] [2022-08-28 12:00:00.000020] created pydantic model
```
### If you must use sync SDK, then run it in a thread pool.
If you must use a library that's not `async`, run the HTTP calls in an external worker thread.
Use `run_in_threadpool` from Starlette.
```python
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
from my_sync_library import SyncAPIClient
app = FastAPI()
@app.get("/")
async def call_my_sync_library():
my_data = await service.get_my_data()
client = SyncAPIClient()
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
# src.profiles.schemas
from pydantic import BaseModel, field_validator
class ProfileCreate(BaseModel):
username: str
password: str
@field_validator("password", mode="after")
@classmethod
def valid_password(cls, password: str) -> str:
if not re.match(STRONG_PASSWORD_PATTERN, password):
raise ValueError(
"Password must contain at least "
"one lower character, "
"one upper character, "
"digit or "
"special symbol"
)
return password
# src.profiles.routes
from fastapi import APIRouter
router = APIRouter()
@router.post("/profiles")
async def create_profile(profile_data: ProfileCreate):
pass
```
**Response Example:**
<img src="images/value_error_response.png" width="400" height="auto">
### Docs
1. Unless your API is public, hide docs by default. Show it explicitly on the selected envs only.
```python
from fastapi import FastAPI
from starlette.config import Config
config = Config(".env") # parse .env file for env variables
ENVIRONMENT = config("ENVIRONMENT") # get current env name
SHOW_DOCS_ENVIRONMENT = ("local", "staging") # explicit list of allowed envs
app_configs = {"title": "My Cool API"}
if ENVIRONMENT not in SHOW_DOCS_ENVIRONMENT:
app_configs["openapi_url"] = None # set url for docs as null
app = FastAPI(**app_configs)
```
2. Help FastAPI to generate an easy-to-understand docs
1. Set `response_model`, `status_code`, `description`, etc.
2. If models and statuses vary, use `responses` route attribute to add docs for different responses
```python
from fastapi import APIRouter, status
router = APIRouter()
@router.post(
"/endpoints",
response_model=DefaultResponseModel, # default response pydantic model
status_code=status.HTTP_201_CREATED, # default status code
description="Description of the well documented endpoint",
tags=["Endpoint Category"],
summary="Summary of the Endpoint",
responses={
status.HTTP_200_OK: {
"model": OkResponse, # custom pydantic model for 200 response
"description": "Ok Response",
},
status.HTTP_201_CREATED: {
"model": CreatedResponse, # custom pydantic model for 201 response
"description": "Creates something from user request",
},
status.HTTP_202_ACCEPTED: {
"model": AcceptedResponse, # custom pydantic model for 202 response
"description": "Accepts request and handles it later",
},
},
)
async def documented_route():
pass
```
Will generate docs like this:
![FastAPI Generated Custom Response Docs](images/custom_responses.png "Custom Response Docs")
### Set DB keys naming conventions
Explicitly setting the indexes' namings according to your database's convention is preferable over sqlalchemy's.
```python
from sqlalchemy import MetaData
POSTGRES_INDEXES_NAMING_CONVENTION = {
"ix": "%(column_0_label)s_idx",
"uq": "%(table_name)s_%(column_0_name)s_key",
"ck": "%(table_name)s_%(constraint_name)s_check",
"fk": "%(table_name)s_%(column_0_name)s_fkey",
"pk": "%(table_name)s_pkey",
}
metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)
```
### Migrations. Alembic
1. Migrations must be static and reversible. If your migrations depend on dynamically generated data, make sure only the data itself is dynamic, not its structure.
2. Generate migrations with descriptive names and slugs. The slug is required and should explain the changes.
3. Set a human-readable file template for new migrations. We use the `*date*_*slug*.py` pattern, e.g., `2022-08-24_post_content_idx.py`
```
# alembic.ini
file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s
```
### Set DB naming conventions
Being consistent with names is important. Some rules we followed:
1. lower_case_snake
2. singular form (e.g. `post`, `post_like`, `user_playlist`)
3. group similar tables with module prefix, e.g. `payment_account`, `payment_bill`, `post`, `post_like`
4. stay consistent across tables, but concrete namings are ok, e.g.
1. use `profile_id` in all tables, but if some of them need only profiles that are creators, use `creator_id`
2. use `post_id` for all abstract tables like `post_like`, `post_view`, but use concrete naming in relevant modules like `course_id` in `chapters.course_id`
5. `_at` suffix for datetime
6. `_date` suffix for date
### SQL-first. Pydantic-second
- 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
from pydantic import UUID4
from sqlalchemy import desc, func, select, text
from sqlalchemy.sql.functions import coalesce
from src.database import database, posts, profiles, post_review, products
async def get_posts(
creator_id: UUID4, *, limit: int = 10, offset: int = 0
) -> list[dict[str, Any]]:
select_query = (
select(
(
posts.c.id,
posts.c.slug,
posts.c.title,
func.json_build_object(
text("'id', profiles.id"),
text("'first_name', profiles.first_name"),
text("'last_name', profiles.last_name"),
text("'username', profiles.username"),
).label("creator"),
)
)
.select_from(posts.join(profiles, posts.c.owner_id == profiles.c.id))
.where(posts.c.owner_id == creator_id)
.limit(limit)
.offset(offset)
.group_by(
posts.c.id,
posts.c.type,
posts.c.slug,
posts.c.title,
profiles.c.id,
profiles.c.first_name,
profiles.c.last_name,
profiles.c.username,
profiles.c.avatar,
)
.order_by(
desc(coalesce(posts.c.updated_at, posts.c.published_at, posts.c.created_at))
)
)
return await database.fetch_all(select_query)
# src.posts.schemas
from typing import Any
from pydantic import BaseModel, UUID4
class Creator(BaseModel):
id: UUID4
first_name: str
last_name: str
username: str
class Post(BaseModel):
id: UUID4
slug: str
title: str
creator: Creator
# src.posts.router
from fastapi import APIRouter, Depends
router = APIRouter()
@router.get("/creators/{creator_id}/posts", response_model=list[Post])
async def get_creator_posts(creator: dict[str, Any] = Depends(valid_creator_id)):
posts = await service.get_posts(creator["id"])
return posts
```
### 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, 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 httpx import AsyncClient, ASGITransport
from src.main import app # inited FastAPI app
@pytest.fixture
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: 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
With linters, you can forget about formatting the code and focus on writing the business logic.
[Ruff](https://github.com/astral-sh/ruff) is "blazingly-fast" new linter that replaces black, autoflake, isort, and supports more than 600 lint rules.
It's a popular good practice to use pre-commit hooks, but just using the script was ok for us.
```shell
#!/bin/sh -e
set -x
ruff check --fix src
ruff format src
```
## Bonus Section
Some very kind people shared their own experience and best practices that are definitely worth reading.
Check them out at [issues](https://github.com/zhanymkanov/fastapi-best-practices/issues) section of the project.
For instance, [lowercase00](https://github.com/zhanymkanov/fastapi-best-practices/issues/4)
has described in details their best practices working with permissions & auth, class-based services & views,
task queues, custom response serializers, configuration with dynaconf, etc.
If you have something to share about your experience working with FastAPI, whether it's good or bad,
you are very welcome to create a new issue. It is our pleasure to read it.

38
alembic.ini Normal file
View File

@@ -0,0 +1,38 @@
[alembic]
script_location = alembic
prepend_sys_path = .
# sqlalchemy.url 由 alembic/env.py 从 src.config.settings 注入，不在此硬编码
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

53
alembic/env.py Normal file
View File

@@ -0,0 +1,53 @@
"""Alembic async env：仅用于 MySQL 后端。MongoDB 无 schema 迁移概念。"""
import asyncio
from logging.config import fileConfig
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# 从应用配置注入 DSN + metadata
import src.items.mysql
from src.config import settings
from src.database import Base
config.set_main_option("sqlalchemy.url", settings.MYSQL_DSN)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
context.configure(
url=config.get_main_option("sqlalchemy.url"),
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def do_run_migrations(connection) -> None:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
async def run_migrations_online() -> None:
connectable = async_engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
if context.is_offline_mode():
run_migrations_offline()
else:
asyncio.run(run_migrations_online())

24
alembic/script.py.mako Normal file
View File

@@ -0,0 +1,24 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
revision: str = ${repr(up_revision)}
down_revision: str | None = ${repr(down_revision)}
branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
depends_on: str | Sequence[str] | None = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

36
alembic/versions/81076765be75_init_items_table.py Normal file
View File

@@ -0,0 +1,36 @@
"""init items table
Revision ID: 81076765be75
Revises:
Create Date: 2026-08-21 21:21:49.098436
"""
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
revision: str = '81076765be75'
down_revision: str | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('items',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('name', sa.String(length=128), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_items_name'), 'items', ['name'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_items_name'), table_name='items')
op.drop_table('items')
# ### end Alembic commands ###

3
alembic/versions/README.md Normal file
View File

@@ -0,0 +1,3 @@
# 迁移版本文件目录。
# 生成迁移：uv run alembic revision --autogenerate -m "add xxx table"
# 执行迁移：uv run alembic upgrade head

34
docker-compose.yml Normal file
View File

@@ -0,0 +1,34 @@
services:
app:
build: .
ports:
- "8000:8000"
env_file: .env
environment:
# compose 内网地址覆盖 .env 里的 127.0.0.1
MYSQL_DSN: mysql+aiomysql://root:root@mysql:3306/app?charset=utf8mb4
MONGO_DSN: mongodb://mongo:27017
depends_on:
- mysql
- mongo
mysql:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: root
MYSQL_DATABASE: app
ports:
- "3306:3306"
volumes:
- mysql_data:/var/lib/mysql
mongo:
image: mongo:8.0
ports:
- "27017:27017"
volumes:
- mongo_data:/data/db
volumes:
mysql_data:
mongo_data:

447
docs/AGENTS_GENERIC.md Normal file
View File

@@ -0,0 +1,447 @@
# FastAPI Best Practices for AI Agents
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 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 # SQLAlchemy ORM models
│ ├── service.py # Business logic
│ ├── dependencies.py # Route dependencies
│ ├── config.py # Domain-scoped BaseSettings
│ ├── constants.py # Constants and error codes
│ ├── exceptions.py # Domain-specific exceptions
│ └── utils.py # Helper functions
├── config.py # Global BaseSettings
├── models.py # Shared Pydantic / ORM bases
├── exceptions.py # Global exceptions
├── database.py # Async engine + session factory
└── main.py # FastAPI app + lifespan
```
**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
### 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
```python
# DON'T — blocking call inside async route freezes the entire event loop
@router.get("/bad")
async def bad():
time.sleep(10) # blocks every request on this worker
return {"ok": True}
# 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}
# 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}
# DO — async route that has to call a sync library
from fastapi.concurrency import run_in_threadpool
@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
```python
from enum import StrEnum
from pydantic import AnyUrl, BaseModel, EmailStr, Field
class MusicBand(StrEnum):
AEROSMITH = "AEROSMITH"
QUEEN = "QUEEN"
ACDC = "AC/DC"
class UserCreate(BaseModel):
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) # required, must be >= 18
favorite_band: MusicBand | None = None
website: AnyUrl | None = None
```
> **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 datetime import datetime
from zoneinfo import ZoneInfo
from pydantic import BaseModel, ConfigDict, field_serializer
class CustomModel(BaseModel):
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
`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_MINUTES: int = 5
REFRESH_TOKEN_KEY: str
REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)
SECURE_COOKIES: bool = True
auth_settings = AuthConfig()
```
## Dependencies
### Use Annotated, not default-arg `Depends(...)`
`Annotated[T, Depends(...)]` is the idiomatic form since FastAPI 0.95 and avoids
gotchas with default values.
```python
# 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
```
### Chain dependencies for reuse
```python
async def valid_owned_post(
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
```
### 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}`).
## Authentication — JWT
Use **`PyJWT`**, not `python-jose` (unmaintained).
```python
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 — SQLAlchemy 2.0 async
Prefer SQLAlchemy 2.0's async API. `encode/databases` is in maintenance mode — don't pick it for new projects.
```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",
"ck": "%(table_name)s_%(constraint_name)s_check",
"fk": "%(table_name)s_%(column_0_name)s_fkey",
"pk": "%(table_name)s_pkey",
}
metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)
```
### 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.
## Background work — BackgroundTasks vs Celery
| 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 |
```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
```
> 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
### Async client from day one
```python
import pytest
from httpx import AsyncClient, ASGITransport
from src.main import app
@pytest.fixture
async def client():
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: 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
```shell
ruff check --fix src
ruff format src
```
Add to a pre-commit hook or run in CI. Ruff replaces black + isort + autoflake + most of flake8.
---
## 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` |

871
docs/BEST_PRACTICES.md Normal file
View File

@@ -0,0 +1,871 @@
## FastAPI Best Practices <!-- omit from toc -->
Opinionated list of best practices and conventions we use at our startups.
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.
*[简体中文](./BEST_PRACTICES_ZH.md)*
## Contents <!-- omit from toc -->
- [Project Structure](#project-structure)
- [Async Routes](#async-routes)
- [I/O Intensive Tasks](#io-intensive-tasks)
- [CPU Intensive Tasks](#cpu-intensive-tasks)
- [Pydantic](#pydantic)
- [Excessively use Pydantic](#excessively-use-pydantic)
- [Custom Base Model](#custom-base-model)
- [Decouple Pydantic BaseSettings](#decouple-pydantic-basesettings)
- [Dependencies](#dependencies)
- [Beyond Dependency Injection](#beyond-dependency-injection)
- [Chain Dependencies](#chain-dependencies)
- [Decouple \& Reuse dependencies. Dependency calls are cached](#decouple--reuse-dependencies-dependency-calls-are-cached)
- [Prefer `async` dependencies](#prefer-async-dependencies)
- [Miscellaneous](#miscellaneous)
- [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)
- [Migrations. Alembic](#migrations-alembic)
- [Set DB naming conventions](#set-db-naming-conventions)
- [SQL-first. Pydantic-second](#sql-first-pydantic-second)
- [Set tests client async from day 0](#set-tests-client-async-from-day-0)
- [Use ruff](#use-ruff)
- [Bonus Section](#bonus-section)
## Project Structure
There are many ways to structure a project, but the best structure is one that is consistent, straightforward, and free of surprises.
Many example projects and tutorials organize projects by file type (e.g., crud, routers, models), which works well for microservices or smaller projects. However, this approach didn't scale well for our monolith with many domains and modules.
The structure I found more scalable and evolvable is inspired by Netflix's [Dispatch](https://github.com/Netflix/dispatch), with some minor modifications.
```
fastapi-project
├── alembic/
├── src
│ ├── auth
│ │ ├── router.py
│ │ ├── schemas.py # pydantic models
│ │ ├── models.py # db models
│ │ ├── dependencies.py
│ │ ├── config.py # local configs
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ ├── service.py
│ │ └── utils.py
│ ├── aws
│ │ ├── client.py # client model for external service communication
│ │ ├── schemas.py
│ │ ├── config.py
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ └── utils.py
│ ├── posts
│ │ ├── router.py
│ │ ├── schemas.py
│ │ ├── models.py
│ │ ├── dependencies.py
│ │ ├── constants.py
│ │ ├── exceptions.py
│ │ ├── service.py
│ │ └── utils.py
│ ├── config.py # global configs
│ ├── models.py # global models
│ ├── exceptions.py # global exceptions
│ ├── pagination.py # global module e.g. pagination
│ ├── database.py # db connection related stuff
│ └── main.py
├── tests/
│ ├── auth
│ ├── aws
│ └── posts
├── templates/
│ └── index.html
├── requirements
│ ├── base.txt
│ ├── dev.txt
│ └── prod.txt
├── .env
├── .gitignore
├── logging.ini
└── alembic.ini
```
1. Store all domain directories inside `src` folder
1. `src/` - highest level of an app, contains common models, configs, and constants, etc.
2. `src/main.py` - root of the project, which inits the FastAPI app
2. Each package has its own router, schemas, models, etc.
1. `router.py` - is a core of each module with all the endpoints
2. `schemas.py` - for pydantic models
3. `models.py` - for db models
4. `service.py` - module specific business logic
5. `dependencies.py` - router dependencies
6. `constants.py` - module specific constants and error codes
7. `config.py` - e.g. env vars
8. `utils.py` - non-business logic functions, e.g. response normalization, data enrichment, etc.
9. `exceptions.py` - module specific exceptions, e.g. `PostNotFound`, `InvalidUserData`
3. When package requires services or dependencies or constants from other packages - import them with an explicit module name
```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 # in case we have Standard ErrorCode in constants module of each package
```
## Async Routes
FastAPI is an async-first framework—it's designed to work with async I/O operations, which is why it's so fast.
However, FastAPI doesn't restrict you to only `async` routes; you can use `sync` routes too. This might confuse beginners into thinking they're the same, but they're not.
### I/O Intensive Tasks
Under the hood, FastAPI can [effectively handle](https://fastapi.tiangolo.com/async/#path-operation-functions) both async and sync I/O operations:
- FastAPI runs `sync` routes in a [threadpool](https://en.wikipedia.org/wiki/Thread_pool), so blocking I/O operations won't stop the [event loop](https://docs.python.org/3/library/asyncio-eventloop.html) from executing other tasks.
- If the route is defined as `async`, it's called via `await` and FastAPI trusts you to only perform non-blocking I/O operations.
The caveat is that if you violate that trust and execute blocking operations within async routes, the event loop won't be able to run other tasks until the blocking operation completes.
```python
import asyncio
import time
from fastapi import APIRouter
router = APIRouter()
@router.get("/terrible-ping")
async def terrible_ping():
time.sleep(10) # I/O blocking operation for 10 seconds, the whole process will be blocked
return {"pong": True}
@router.get("/good-ping")
def good_ping():
time.sleep(10) # I/O blocking operation for 10 seconds, but in a separate thread for the whole `good_ping` route
return {"pong": True}
@router.get("/perfect-ping")
async def perfect_ping():
await asyncio.sleep(10) # non-blocking I/O operation
return {"pong": True}
```
**What happens when we call:**
1. `GET /terrible-ping`
1. FastAPI server receives a request and starts handling it
2. Server's event loop and all queued tasks wait until `time.sleep()` finishes
1. Since the route is `async`, the server doesn't offload it to a threadpool—it blocks the entire event loop
2. Server won't accept any new requests while waiting
3. Server returns the response
1. Only after responding does the server resume accepting new requests
2. `GET /good-ping`
1. FastAPI server receives a request and starts handling it
2. FastAPI sends the entire `good_ping` route to the threadpool, where a worker thread runs the function
3. While `good_ping` executes, the event loop continues processing other tasks (e.g., accepting new requests, calling the database)
- The worker thread waits for `time.sleep` to finish, independently of the main thread
- The sync operation blocks only the worker thread, not the main event loop
4. When `good_ping` finishes, the server returns a response to the client
3. `GET /perfect-ping`
1. FastAPI server receives a request and starts handling it
2. FastAPI awaits `asyncio.sleep(10)`
3. Event loop continues processing other tasks from the queue (e.g., accepting new requests, calling the database)
4. When `asyncio.sleep(10)` completes, the server finishes executing the route and returns a response to the client
> [!WARNING]
> Notes on the thread pool:
> - Threads require more resources than coroutines, so they are not as cheap as async I/O operations.
> - Thread pool has a limited number of threads, i.e. you might run out of threads and your app will become slow. [Read more](https://github.com/Kludex/fastapi-tips?tab=readme-ov-file#2-be-careful-with-non-async-functions) (external link)
### CPU Intensive Tasks
The second caveat is that non-blocking awaitables and threadpool offloading are only beneficial for I/O intensive tasks (e.g., file operations, database calls, external API requests).
- Awaiting CPU-intensive tasks (e.g., heavy calculations, data processing, video transcoding) provides no benefit since the CPU must actively work to complete them. In contrast, I/O operations are external—the server just waits for a response and can handle other tasks in the meantime.
- Running CPU-intensive tasks in other threads is also ineffective due to the [GIL](https://realpython.com/python-gil/). In short, the GIL allows only one thread to execute Python bytecode at a time, making threads ineffective for CPU-bound work.
- To optimize CPU-intensive tasks, you should offload them to worker processes (e.g., using `multiprocessing` or a task queue like Celery).
**Related StackOverflow questions of confused users**
1. https://stackoverflow.com/questions/62976648/architecture-flask-vs-fastapi/70309597#70309597
- Here you can also check [my answer](https://stackoverflow.com/a/70309597/6927498)
2. https://stackoverflow.com/questions/65342833/fastapi-uploadfile-is-slow-compared-to-flask
3. https://stackoverflow.com/questions/71516140/fastapi-runs-api-calls-in-serial-instead-of-parallel-fashion
## Pydantic
### Excessively use Pydantic
Pydantic has a rich set of features to validate and transform data.
In addition to standard features like required and optional fields with default values,
Pydantic has built-in data processing tools like regex validation, enums, string manipulation, email validation, and more.
```python
from enum import StrEnum
from pydantic import AnyUrl, BaseModel, EmailStr, Field
class MusicBand(StrEnum):
AEROSMITH = "AEROSMITH"
QUEEN = "QUEEN"
ACDC = "AC/DC"
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) # 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
```
### Custom Base Model
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, field_serializer
class CustomModel(BaseModel):
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."""
default_dict = self.model_dump()
return jsonable_encoder(default_dict)
```
In the example above, we have decided to create a global base model that:
- Serializes all datetime fields to a standard format with an explicit timezone
- Provides a method to return a dict with only serializable fields
### Decouple Pydantic BaseSettings
BaseSettings is great for reading environment variables, but a single BaseSettings for the whole app gets messy. Split it across modules and domains.
```python
# src.auth.config
from datetime import timedelta
from pydantic_settings import BaseSettings
class AuthConfig(BaseSettings):
JWT_ALG: str
JWT_SECRET: str
JWT_EXP: int = 5 # minutes
REFRESH_TOKEN_KEY: str
REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)
SECURE_COOKIES: bool = True
auth_settings = AuthConfig()
# src.config
from pydantic import PostgresDsn, RedisDsn
from pydantic_settings import BaseSettings
from src.constants import Environment
class Config(BaseSettings):
DATABASE_URL: PostgresDsn
REDIS_URL: RedisDsn
SITE_DOMAIN: str = "myapp.com"
ENVIRONMENT: Environment = Environment.PRODUCTION
SENTRY_DSN: str | None = None
CORS_ORIGINS: list[str]
CORS_ORIGINS_REGEX: str | None = None
CORS_HEADERS: list[str]
APP_VERSION: str = "1.0"
settings = Config()
```
## Dependencies
### Beyond Dependency Injection
Pydantic is a great schema validator, but for complex validations that require database or external service calls, it's not enough.
FastAPI docs mostly present dependencies as DI for endpoints, but they're also great for request validation.
Dependencies can validate data against database constraints (e.g., checking if an email already exists, ensuring a user exists, etc.).
```python
# dependencies.py
async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
post = await service.get_by_id(post_id)
if not post:
raise PostNotFound()
return post
# router.py
@router.get("/posts/{post_id}", response_model=PostResponse)
async def get_post_by_id(post: dict[str, Any] = Depends(valid_post_id)):
return post
@router.put("/posts/{post_id}", response_model=PostResponse)
async def update_post(
update_data: PostUpdate,
post: dict[str, Any] = Depends(valid_post_id),
):
updated_post = await service.update(id=post["id"], data=update_data)
return updated_post
@router.get("/posts/{post_id}/reviews", response_model=list[ReviewsResponse])
async def get_post_reviews(post: dict[str, Any] = Depends(valid_post_id)):
post_reviews = await reviews_service.get_by_post_id(post["id"])
return post_reviews
```
If we didn't put data validation in a dependency, we would have to validate that `post_id` exists
in every endpoint and write the same tests for each of them.
### Chain Dependencies
Dependencies can use other dependencies and avoid code repetition for similar logic.
```python
# dependencies.py
from fastapi.security import OAuth2PasswordBearer
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)
if not post:
raise PostNotFound()
return post
async def parse_jwt_data(
token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict[str, Any]:
try:
payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
except InvalidTokenError:
raise InvalidCredentials()
return {"user_id": payload["id"]}
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]:
if post["creator_id"] != token_data["user_id"]:
raise UserNotOwner()
return post
# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(post: dict[str, Any] = Depends(valid_owned_post)):
return post
```
### Decouple & Reuse dependencies. Dependency calls are cached
Dependencies can be reused multiple times, and they won't be recalculated - FastAPI caches dependency's result within a request's scope by default,
i.e. if `valid_post_id` gets called multiple times in one route, it will be called only once.
Knowing this, we can decouple dependencies onto multiple smaller functions that operate on a smaller domain and are easier to reuse in other routes.
For example, in the code below we are using `parse_jwt_data` three times:
1. `valid_owned_post`
2. `valid_active_creator`
3. `get_user_post`,
but `parse_jwt_data` is called only once, in the very first call.
```python
# dependencies.py
from fastapi import BackgroundTasks
from fastapi.security import OAuth2PasswordBearer
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)
if not post:
raise PostNotFound()
return post
async def parse_jwt_data(
token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict:
try:
payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
except InvalidTokenError:
raise InvalidCredentials()
return {"user_id": payload["id"]}
async def valid_owned_post(
post: Mapping = Depends(valid_post_id),
token_data: dict = Depends(parse_jwt_data),
) -> Mapping:
if post["creator_id"] != token_data["user_id"]:
raise UserNotOwner()
return post
async def valid_active_creator(
token_data: dict = Depends(parse_jwt_data),
):
user = await users_service.get_by_id(token_data["user_id"])
if not user["is_active"]:
raise UserIsBanned()
if not user["is_creator"]:
raise UserNotCreator()
return user
# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(
worker: BackgroundTasks,
post: Mapping = Depends(valid_owned_post),
user: Mapping = Depends(valid_active_creator),
):
"""Get post that belong the active user."""
worker.add_task(notifications_service.send_email, user["id"])
return post
```
### Prefer `async` dependencies
FastAPI supports both `sync` and `async` dependencies. It's tempting to use `sync` when you don't need to await anything, but that's not the best choice.
Just like routes, `sync` dependencies run in a threadpool. Threads have overhead that's unnecessary for small non-I/O operations.
[See more](https://github.com/Kludex/fastapi-tips?tab=readme-ov-file#9-your-dependencies-may-be-running-on-threads) (external link)
## Miscellaneous
### Follow the REST
Developing RESTful API makes it easier to reuse dependencies in routes like these:
1. `GET /courses/:course_id`
2. `GET /courses/:course_id/chapters/:chapter_id/lessons`
3. `GET /chapters/:chapter_id`
The only caveat is having to use the same variable names in the path:
- If you have two endpoints `GET /profiles/:profile_id` and `GET /creators/:creator_id`
that both validate whether the given `profile_id` exists, but `GET /creators/:creator_id`
also checks if the profile is creator, then it's better to rename `creator_id` path variable to `profile_id` and chain those two dependencies.
```python
# src.profiles.dependencies
async def valid_profile_id(profile_id: UUID4) -> Mapping:
profile = await service.get_by_id(profile_id)
if not profile:
raise ProfileNotFound()
return profile
# src.creators.dependencies
async def valid_creator_id(profile: Mapping = Depends(valid_profile_id)) -> Mapping:
if not profile["is_creator"]:
raise ProfileNotCreator()
return profile
# src.profiles.router.py
@router.get("/profiles/{profile_id}", response_model=ProfileResponse)
async def get_user_profile_by_id(profile: Mapping = Depends(valid_profile_id)):
"""Get profile by id."""
return profile
# src.creators.router.py
@router.get("/creators/{profile_id}", response_model=ProfileResponse)
async def get_user_profile_by_id(
creator_profile: Mapping = Depends(valid_creator_id)
):
"""Get creator's profile by id."""
return creator_profile
```
### FastAPI response serialization
You might think you can return a Pydantic object that matches your route's `response_model` and skip some processing steps, but you'd be wrong.
FastAPI first converts the Pydantic object to a dict using `jsonable_encoder`, then validates the data against your `response_model`, and only then serializes it to JSON.
This means your Pydantic model object is created twice:
- First, when you explicitly create it to return from your route.
- Second, implicitly by FastAPI to validate the response data according to the response_model.
```python
from fastapi import FastAPI
from pydantic import BaseModel, model_validator
app = FastAPI()
class ProfileResponse(BaseModel):
@model_validator(mode="after")
def debug_usage(self):
print("created pydantic model")
return self
@app.get("/", response_model=ProfileResponse)
async def root():
return ProfileResponse()
```
**Logs Output:**
```
[INFO] [2022-08-28 12:00:00.000000] created pydantic model
[INFO] [2022-08-28 12:00:00.000020] created pydantic model
```
### If you must use sync SDK, then run it in a thread pool.
If you must use a library that's not `async`, run the HTTP calls in an external worker thread.
Use `run_in_threadpool` from Starlette.
```python
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
from my_sync_library import SyncAPIClient
app = FastAPI()
@app.get("/")
async def call_my_sync_library():
my_data = await service.get_my_data()
client = SyncAPIClient()
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
# src.profiles.schemas
from pydantic import BaseModel, field_validator
class ProfileCreate(BaseModel):
username: str
password: str
@field_validator("password", mode="after")
@classmethod
def valid_password(cls, password: str) -> str:
if not re.match(STRONG_PASSWORD_PATTERN, password):
raise ValueError(
"Password must contain at least "
"one lower character, "
"one upper character, "
"digit or "
"special symbol"
)
return password
# src.profiles.routes
from fastapi import APIRouter
router = APIRouter()
@router.post("/profiles")
async def create_profile(profile_data: ProfileCreate):
pass
```
**Response Example:**
<img src="images/value_error_response.png" width="400" height="auto">
### Docs
1. Unless your API is public, hide docs by default. Show it explicitly on the selected envs only.
```python
from fastapi import FastAPI
from starlette.config import Config
config = Config(".env") # parse .env file for env variables
ENVIRONMENT = config("ENVIRONMENT") # get current env name
SHOW_DOCS_ENVIRONMENT = ("local", "staging") # explicit list of allowed envs
app_configs = {"title": "My Cool API"}
if ENVIRONMENT not in SHOW_DOCS_ENVIRONMENT:
app_configs["openapi_url"] = None # set url for docs as null
app = FastAPI(**app_configs)
```
2. Help FastAPI to generate an easy-to-understand docs
1. Set `response_model`, `status_code`, `description`, etc.
2. If models and statuses vary, use `responses` route attribute to add docs for different responses
```python
from fastapi import APIRouter, status
router = APIRouter()
@router.post(
"/endpoints",
response_model=DefaultResponseModel, # default response pydantic model
status_code=status.HTTP_201_CREATED, # default status code
description="Description of the well documented endpoint",
tags=["Endpoint Category"],
summary="Summary of the Endpoint",
responses={
status.HTTP_200_OK: {
"model": OkResponse, # custom pydantic model for 200 response
"description": "Ok Response",
},
status.HTTP_201_CREATED: {
"model": CreatedResponse, # custom pydantic model for 201 response
"description": "Creates something from user request",
},
status.HTTP_202_ACCEPTED: {
"model": AcceptedResponse, # custom pydantic model for 202 response
"description": "Accepts request and handles it later",
},
},
)
async def documented_route():
pass
```
Will generate docs like this:
![FastAPI Generated Custom Response Docs](images/custom_responses.png "Custom Response Docs")
### Set DB keys naming conventions
Explicitly setting the indexes' namings according to your database's convention is preferable over sqlalchemy's.
```python
from sqlalchemy import MetaData
POSTGRES_INDEXES_NAMING_CONVENTION = {
"ix": "%(column_0_label)s_idx",
"uq": "%(table_name)s_%(column_0_name)s_key",
"ck": "%(table_name)s_%(constraint_name)s_check",
"fk": "%(table_name)s_%(column_0_name)s_fkey",
"pk": "%(table_name)s_pkey",
}
metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)
```
### Migrations. Alembic
1. Migrations must be static and reversible. If your migrations depend on dynamically generated data, make sure only the data itself is dynamic, not its structure.
2. Generate migrations with descriptive names and slugs. The slug is required and should explain the changes.
3. Set a human-readable file template for new migrations. We use the `*date*_*slug*.py` pattern, e.g., `2022-08-24_post_content_idx.py`
```
# alembic.ini
file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s
```
### Set DB naming conventions
Being consistent with names is important. Some rules we followed:
1. lower_case_snake
2. singular form (e.g. `post`, `post_like`, `user_playlist`)
3. group similar tables with module prefix, e.g. `payment_account`, `payment_bill`, `post`, `post_like`
4. stay consistent across tables, but concrete namings are ok, e.g.
1. use `profile_id` in all tables, but if some of them need only profiles that are creators, use `creator_id`
2. use `post_id` for all abstract tables like `post_like`, `post_view`, but use concrete naming in relevant modules like `course_id` in `chapters.course_id`
5. `_at` suffix for datetime
6. `_date` suffix for date
### SQL-first. Pydantic-second
- 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
from pydantic import UUID4
from sqlalchemy import desc, func, select, text
from sqlalchemy.sql.functions import coalesce
from src.database import database, posts, profiles, post_review, products
async def get_posts(
creator_id: UUID4, *, limit: int = 10, offset: int = 0
) -> list[dict[str, Any]]:
select_query = (
select(
(
posts.c.id,
posts.c.slug,
posts.c.title,
func.json_build_object(
text("'id', profiles.id"),
text("'first_name', profiles.first_name"),
text("'last_name', profiles.last_name"),
text("'username', profiles.username"),
).label("creator"),
)
)
.select_from(posts.join(profiles, posts.c.owner_id == profiles.c.id))
.where(posts.c.owner_id == creator_id)
.limit(limit)
.offset(offset)
.group_by(
posts.c.id,
posts.c.type,
posts.c.slug,
posts.c.title,
profiles.c.id,
profiles.c.first_name,
profiles.c.last_name,
profiles.c.username,
profiles.c.avatar,
)
.order_by(
desc(coalesce(posts.c.updated_at, posts.c.published_at, posts.c.created_at))
)
)
return await database.fetch_all(select_query)
# src.posts.schemas
from typing import Any
from pydantic import BaseModel, UUID4
class Creator(BaseModel):
id: UUID4
first_name: str
last_name: str
username: str
class Post(BaseModel):
id: UUID4
slug: str
title: str
creator: Creator
# src.posts.router
from fastapi import APIRouter, Depends
router = APIRouter()
@router.get("/creators/{creator_id}/posts", response_model=list[Post])
async def get_creator_posts(creator: dict[str, Any] = Depends(valid_creator_id)):
posts = await service.get_posts(creator["id"])
return posts
```
### 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, 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 httpx import AsyncClient, ASGITransport
from src.main import app # inited FastAPI app
@pytest.fixture
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: 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
With linters, you can forget about formatting the code and focus on writing the business logic.
[Ruff](https://github.com/astral-sh/ruff) is "blazingly-fast" new linter that replaces black, autoflake, isort, and supports more than 600 lint rules.
It's a popular good practice to use pre-commit hooks, but just using the script was ok for us.
```shell
#!/bin/sh -e
set -x
ruff check --fix src
ruff format src
```
## Bonus Section
Some very kind people shared their own experience and best practices that are definitely worth reading.
Check them out at [issues](https://github.com/zhanymkanov/fastapi-best-practices/issues) section of the project.
For instance, [lowercase00](https://github.com/zhanymkanov/fastapi-best-practices/issues/4)
has described in details their best practices working with permissions & auth, class-based services & views,
task queues, custom response serializers, configuration with dynaconf, etc.
If you have something to share about your experience working with FastAPI, whether it's good or bad,
you are very welcome to create a new issue. It is our pleasure to read it.

0
README_ZH.md → docs/BEST_PRACTICES_ZH.md
View File

0
images/custom_responses.png → docs/images/custom_responses.png
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 684 KiB

After

Width:  |  Height:  |  Size: 684 KiB

0
images/type_hints.png → docs/images/type_hints.png
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 390 KiB

After

Width:  |  Height:  |  Size: 390 KiB

0
images/type_hintsless.png → docs/images/type_hintsless.png
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 330 KiB

After

Width:  |  Height:  |  Size: 330 KiB

0
images/value_error_response.png → docs/images/value_error_response.png
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 126 KiB

After

Width:  |  Height:  |  Size: 126 KiB

50
pyproject.toml Normal file
View File

@@ -0,0 +1,50 @@
[project]
name = "fastapi-template"
version = "0.1.0"
description = "开箱即用的 FastAPI 模板：MySQL(SQLAlchemy2.0 async) / MongoDB(Beanie) 可切换"
requires-python = ">=3.11"
dependencies = [
"fastapi>=0.115",
"uvicorn[standard]>=0.30",
"pydantic>=2.7",
"pydantic-settings>=2.4",
# MySQL backend
"sqlalchemy[asyncio]>=2.0",
"aiomysql>=0.2.0",
"alembic>=1.13",
# MongoDB backend
"beanie>=2.0",
"pymongo>=4.9",
"cryptography>=50.0.0",
]
[dependency-groups]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23",
"httpx>=0.27",
"ruff>=0.6",
]
[tool.ruff]
line-length = 100
target-version = "py311"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "ASYNC", "RUF"]
ignore = [
"RUF001", "RUF002", "RUF003", # 中文文档字符串里的全角标点是有意的
]
[tool.ruff.lint.per-file-ignores]
"alembic/env.py" = ["E402", "F401"] # 需要先 fileConfig 再 import 应用模块
"alembic/versions/*" = ["I001"] # 迁移文件是自动生成物，不整理 import
"tests/conftest.py" = ["E402"] # 需要先设环境变量再 import app
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pythonpath = ["."]
[tool.uv]
package = false

0
src/__init__.py Normal file
View File

39
src/config.py Normal file
View File

@@ -0,0 +1,39 @@
"""全局配置：pydantic-settings，环境变量 / .env 驱动。"""
from enum import StrEnum
from pydantic_settings import BaseSettings, SettingsConfigDict
class DBBackend(StrEnum):
MYSQL = "mysql"
MONGODB = "mongodb"
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
# 数据库后端开关：mysql / mongodb
DB_BACKEND: DBBackend = DBBackend.MONGODB
# MySQL（SQLAlchemy async DSN）
MYSQL_DSN: str = "mysql+aiomysql://root:root@127.0.0.1:3306/app?charset=utf8mb4"
# MongoDB
MONGO_DSN: str = "mongodb://127.0.0.1:27017"
MONGO_DB: str = "app"
# 应用
APP_NAME: str = "fastapi-template"
APP_ENV: str = "dev"
APP_DEBUG: bool = True
APP_HOST: str = "0.0.0.0"
APP_PORT: int = 8000
CORS_ORIGINS: str = "*"
@property
def cors_origin_list(self) -> list[str]:
return [o.strip() for o in self.CORS_ORIGINS.split(",") if o.strip()]
settings = Settings()

28
src/database.py Normal file
View File

@@ -0,0 +1,28 @@
"""MySQL：SQLAlchemy 2.0 async engine + session factory。
仅 DB_BACKEND=mysql 时使用；mongodb 模式下本模块不被引用。
"""
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
from src.config import settings
class Base(DeclarativeBase):
"""所有 MySQL ORM 模型的基类。"""
engine = create_async_engine(settings.MYSQL_DSN, pool_size=10, pool_recycle=3600)
async_session = async_sessionmaker(engine, expire_on_commit=False)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
"""FastAPI 依赖：每请求一个 session。"""
async with async_session() as session:
yield session
async def close_engine() -> None:
await engine.dispose()

26
src/exceptions.py Normal file
View File

@@ -0,0 +1,26 @@
"""全局异常 + 统一错误响应。"""
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
class AppError(Exception):
"""业务异常基类。子类在各域 exceptions.py 中定义。"""
status_code: int = status.HTTP_400_BAD_REQUEST
detail: str = "bad request"
def __init__(self, detail: str | None = None):
self.detail = detail or self.detail
super().__init__(self.detail)
class NotFoundError(AppError):
status_code = status.HTTP_404_NOT_FOUND
detail = "not found"
def register_exception_handlers(app: FastAPI) -> None:
@app.exception_handler(AppError)
async def app_error_handler(_: Request, exc: AppError) -> JSONResponse:
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})

0
src/items/__init__.py Normal file
View File

40
src/items/dependencies.py Normal file
View File

@@ -0,0 +1,40 @@
"""items 域依赖：按 DB_BACKEND 选择仓储实现。
router 只依赖 ItemRepoDep；切库只改 .env，业务代码零改动。
"""
from typing import Annotated, Protocol
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from src.config import settings
from src.database import get_db
from src.items.mysql import MySQLItemRepo
from src.items.schemas import ItemCreate, ItemOut, ItemUpdate
class ItemRepo(Protocol):
"""仓储协议：新增域时定义同样的 Protocol，两后端各自实现。"""
async def create(self, data: ItemCreate) -> ItemOut: ...
async def get(self, item_id: str) -> ItemOut | None: ...
async def list(self, skip: int = 0, limit: int = 20) -> tuple[int, list[ItemOut]]: ...
async def update(self, item_id: str, data: ItemUpdate) -> ItemOut | None: ...
async def delete(self, item_id: str) -> bool: ...
async def _get_mysql_repo(session: Annotated[AsyncSession, Depends(get_db)]) -> ItemRepo:
return MySQLItemRepo(session)
async def _get_mongo_repo() -> ItemRepo:
from src.items.mongo import MongoItemRepo
return MongoItemRepo()
# 模块加载时按配置选定实现
get_item_repo = _get_mysql_repo if settings.DB_BACKEND == "mysql" else _get_mongo_repo
ItemRepoDep = Annotated[ItemRepo, Depends(get_item_repo)]

57
src/items/mongo.py Normal file
View File

@@ -0,0 +1,57 @@
"""items 域 MongoDB 实现：Beanie ODM。"""
from datetime import UTC, datetime
from beanie import Document, PydanticObjectId
from pydantic import Field
from src.items.schemas import ItemCreate, ItemOut, ItemUpdate
class ItemDoc(Document):
name: str = Field(max_length=128)
description: str | None = None
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
class Settings:
name = "items"
def to_out(self) -> ItemOut:
return ItemOut(
id=str(self.id),
name=self.name,
description=self.description,
created_at=self.created_at,
)
class MongoItemRepo:
"""与 MySQLItemRepo 同接口，router 无感知切换。"""
async def create(self, data: ItemCreate) -> ItemOut:
doc = ItemDoc(name=data.name, description=data.description)
await doc.insert()
return doc.to_out()
async def get(self, item_id: str) -> ItemOut | None:
doc = await ItemDoc.get(PydanticObjectId(item_id))
return doc.to_out() if doc else None
async def list(self, skip: int = 0, limit: int = 20) -> tuple[int, list[ItemOut]]:
total = await ItemDoc.count()
docs = await ItemDoc.find_all().skip(skip).limit(limit).to_list()
return total, [d.to_out() for d in docs]
async def update(self, item_id: str, data: ItemUpdate) -> ItemOut | None:
doc = await ItemDoc.get(PydanticObjectId(item_id))
if not doc:
return None
await doc.set(data.model_dump(exclude_unset=True))
return doc.to_out()
async def delete(self, item_id: str) -> bool:
doc = await ItemDoc.get(PydanticObjectId(item_id))
if not doc:
return False
await doc.delete()
return True

68
src/items/mysql.py Normal file
View File

@@ -0,0 +1,68 @@
"""items 域 MySQL 实现：SQLAlchemy 2.0 async。"""
from datetime import datetime
from sqlalchemy import String, Text, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, mapped_column
from src.database import Base
from src.items.schemas import ItemCreate, ItemOut, ItemUpdate
class Item(Base):
__tablename__ = "items"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(128), index=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
def to_out(self) -> ItemOut:
return ItemOut(
id=str(self.id),
name=self.name,
description=self.description,
created_at=self.created_at,
)
class MySQLItemRepo:
"""仓储模式：所有 SQL 集中在此，router 不直接碰 ORM。"""
def __init__(self, session: AsyncSession):
self.session = session
async def create(self, data: ItemCreate) -> ItemOut:
item = Item(name=data.name, description=data.description)
self.session.add(item)
await self.session.commit()
await self.session.refresh(item)
return item.to_out()
async def get(self, item_id: str) -> ItemOut | None:
item = await self.session.get(Item, int(item_id))
return item.to_out() if item else None
async def list(self, skip: int = 0, limit: int = 20) -> tuple[int, list[ItemOut]]:
total = await self.session.scalar(select(func.count(Item.id)))
rows = await self.session.scalars(select(Item).offset(skip).limit(limit))
return total or 0, [r.to_out() for r in rows]
async def update(self, item_id: str, data: ItemUpdate) -> ItemOut | None:
item = await self.session.get(Item, int(item_id))
if not item:
return None
for field, value in data.model_dump(exclude_unset=True).items():
setattr(item, field, value)
await self.session.commit()
await self.session.refresh(item)
return item.to_out()
async def delete(self, item_id: str) -> bool:
item = await self.session.get(Item, int(item_id))
if not item:
return False
await self.session.delete(item)
await self.session.commit()
return True

46
src/items/router.py Normal file
View File

@@ -0,0 +1,46 @@
"""items 域路由：只依赖 ItemRepo 协议，不关心底层是 MySQL 还是 MongoDB。"""
from fastapi import APIRouter, Query, status
from src.exceptions import NotFoundError
from src.items.dependencies import ItemRepoDep
from src.items.schemas import ItemCreate, ItemList, ItemOut, ItemUpdate
router = APIRouter(prefix="/api/v1/items", tags=["items"])
@router.post("", response_model=ItemOut, status_code=status.HTTP_201_CREATED)
async def create_item(data: ItemCreate, repo: ItemRepoDep):
return await repo.create(data)
@router.get("/{item_id}", response_model=ItemOut)
async def get_item(item_id: str, repo: ItemRepoDep):
item = await repo.get(item_id)
if not item:
raise NotFoundError(f"item {item_id} not found")
return item
@router.get("", response_model=ItemList)
async def list_items(
repo: ItemRepoDep,
skip: int = Query(default=0, ge=0),
limit: int = Query(default=20, ge=1, le=100),
):
total, items = await repo.list(skip=skip, limit=limit)
return ItemList(total=total, items=items)
@router.patch("/{item_id}", response_model=ItemOut)
async def update_item(item_id: str, data: ItemUpdate, repo: ItemRepoDep):
item = await repo.update(item_id, data)
if not item:
raise NotFoundError(f"item {item_id} not found")
return item
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: str, repo: ItemRepoDep):
if not await repo.delete(item_id):
raise NotFoundError(f"item {item_id} not found")

27
src/items/schemas.py Normal file
View File

@@ -0,0 +1,27 @@
"""items 域 API 模型（对外契约，与 DB 实现无关）。"""
from datetime import datetime
from pydantic import BaseModel, Field
class ItemCreate(BaseModel):
name: str = Field(min_length=1, max_length=128)
description: str | None = Field(default=None, max_length=1024)
class ItemUpdate(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=128)
description: str | None = Field(default=None, max_length=1024)
class ItemOut(BaseModel):
id: str
name: str
description: str | None
created_at: datetime
class ItemList(BaseModel):
total: int
items: list[ItemOut]

64
src/main.py Normal file
View File

@@ -0,0 +1,64 @@
"""应用入口：FastAPI app + lifespan。
数据库按 settings.DB_BACKEND 初始化：
- mysql → 建 engine，启动时自动建表（生产用 alembic，见 README）
- mongodb → motor client + beanie init
"""
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from src.config import DBBackend, settings
from src.exceptions import register_exception_handlers
from src.items.router import router as items_router
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
if settings.DB_BACKEND == DBBackend.MYSQL:
from src.database import Base, close_engine, engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
await close_engine()
else:
from src.mongo import close_mongo, init_mongo
await init_mongo()
yield
await close_mongo()
def create_app() -> FastAPI:
app = FastAPI(
title=settings.APP_NAME,
debug=settings.APP_DEBUG,
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origin_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
register_exception_handlers(app)
app.include_router(items_router)
@app.get("/health", tags=["meta"])
async def health() -> dict:
return {"status": "ok", "db_backend": str(settings.DB_BACKEND)}
return app
app = create_app()
if __name__ == "__main__":
import uvicorn
uvicorn.run("src.main:app", host=settings.APP_HOST, port=settings.APP_PORT, reload=True)

31
src/mongo.py Normal file
View File

@@ -0,0 +1,31 @@
"""MongoDB：pymongo async client + beanie 初始化。
仅 DB_BACKEND=mongodb 时使用；mysql 模式下本模块不被引用。
（beanie 2.x 基于 pymongo 原生异步客户端，不再需要 motor）
"""
from beanie import init_beanie
from pymongo import AsyncMongoClient
from src.config import settings
_client: AsyncMongoClient | None = None
def _collect_documents() -> list[type]:
"""汇总所有 beanie Document。新增域时在此登记。"""
from src.items import mongo as item_mongo
return [item_mongo.ItemDoc]
async def init_mongo() -> None:
global _client
_client = AsyncMongoClient(settings.MONGO_DSN)
await init_beanie(database=_client[settings.MONGO_DB], document_models=_collect_documents())
async def close_mongo() -> None:
global _client
if _client is not None:
await _client.close()
_client = None

30
tests/conftest.py Normal file
View File

@@ -0,0 +1,30 @@
"""测试基座：httpx ASGITransport 进程内测试，不起真实服务。
默认跑 mongodb 后端（需要本地 27017 有 mongo）。
跑 mysql 后端：DB_BACKEND=mysql MYSQL_DSN=... uv run pytest
"""
import os
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
# 测试用独立库，避免污染开发数据
os.environ.setdefault("MONGO_DB", "app_test")
if os.environ.get("DB_BACKEND") == "mysql":
os.environ["MYSQL_DSN"] = os.environ.get(
"MYSQL_DSN", "mysql+aiomysql://root:root@127.0.0.1:3306/app_test?charset=utf8mb4"
)
from src.main import create_app
@pytest_asyncio.fixture
async def client():
app = create_app()
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
# 手动触发 lifespan（ASGITransport 不自动跑 lifespan）
async with app.router.lifespan_context(app):
yield ac

33
tests/test_items.py Normal file
View File

@@ -0,0 +1,33 @@
async def test_health(client):
resp = await client.get("/health")
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
async def test_items_crud(client):
# create
resp = await client.post("/api/v1/items", json={"name": "demo", "description": "d"})
assert resp.status_code == 201
item = resp.json()
assert item["name"] == "demo"
item_id = item["id"]
# get
resp = await client.get(f"/api/v1/items/{item_id}")
assert resp.status_code == 200
# list
resp = await client.get("/api/v1/items")
assert resp.status_code == 200
assert resp.json()["total"] >= 1
# update
resp = await client.patch(f"/api/v1/items/{item_id}", json={"name": "demo2"})
assert resp.status_code == 200
assert resp.json()["name"] == "demo2"
# delete
resp = await client.delete(f"/api/v1/items/{item_id}")
assert resp.status_code == 204
resp = await client.get(f"/api/v1/items/{item_id}")
assert resp.status_code == 404

1321
uv.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff