---

上一篇【第026篇】CLI 工具开发——Click + Rich 打造专业命令行应用
下一篇【第028篇】自动化脚本实战——文件处理、定时任务与 Web 爬虫

---

系列说明：本系列共 30 篇，旨在帮助Python学习者从零基础到精通。本系列强调实战导向，每篇文章都配有可运行的代码示例。本文为第 027 篇，聚焦于 FastAPI Web API 构建。

---

摘要

FastAPI 以 Pydantic v2 数据校验、自动生成 OpenAPI 文档、原生异步支持三大特性成为 Python Web API 开发的首选框架。本文从路由参数解析、请求体校验讲起，深入依赖注入、JWT 认证、异步 SQLAlchemy 2.0，最终构建一个生产可用的 Todo CRUD API，涵盖 httpx 测试、Uvicorn 部署与 Docker 容器化全流程。

---

1. FastAPI 简介与框架对比

FastAPI 由 Sebastián Ramírez（tiangolo）于 2018 年开发，基于 Starlette（ASGI 框架）和 Pydantic（数据验证）构建。

特性	Flask	Django REST	FastAPI
接口类型	WSGI	WSGI	ASGI（原生异步）
数据校验	手动/marshmallow	DRF Serializer	Pydantic v2（自动）
OpenAPI 文档	Flask-RESTX	drf-spectacular	内置自动生成
类型提示	可选	可选	核心设计
性能（请求/秒）	~500	~400	~3000+
学习曲线	低	高	中

安装依赖：

pip install fastapi uvicorn[standard] pydantic[email]
# 数据库相关
pip install sqlalchemy[asyncio] aiosqlite
# 认证相关
pip install pyjwt pwdlib[argon2]
# 测试相关
pip install httpx pytest pytest-asyncio

---

2. 路由与参数

2.1 路径参数与查询参数

from fastapi import FastAPI, Path, Query, HTTPException
from enum import Enum

app = FastAPI(title='Todo API', version='1.0.0')


class SortOrder(str, Enum):
    asc = 'asc'
    desc = 'desc'


@app.get('/items/{item_id}')
async def get_item(
    item_id: int = Path(..., ge=1, le=9999, description='Item ID'),
    include_deleted: bool = Query(False, description='是否包含已删除项'),
) -> dict:
    if item_id == 404:
        raise HTTPException(status_code=404, detail='Item not found')
    return {'id': item_id, 'include_deleted': include_deleted}


@app.get('/items')
async def list_items(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
    sort: SortOrder = Query(SortOrder.asc),
    keyword: str | None = Query(None, min_length=2, max_length=50),
) -> dict:
    offset = (page - 1) * size
    return {'page': page, 'size': size, 'sort': sort, 'offset': offset}

2.2 请求体与 Pydantic v2

from pydantic import BaseModel, Field, field_validator, model_validator
from datetime import datetime


class TodoCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200, description='待办标题')
    description: str | None = Field(None, max_length=2000)
    priority: int = Field(1, ge=1, le=5, description='优先级 1-5')
    due_date: datetime | None = None
    tags: list[str] = Field(default_factory=list, max_length=10)

    @field_validator('title')
    @classmethod
    def title_strip(cls, v: str) -> str:
        """去除首尾空白"""
        v = v.strip()
        if not v:
            raise ValueError('标题不能为空白字符')
        return v

    @field_validator('tags', mode='before')
    @classmethod
    def tags_unique(cls, v: list) -> list:
        """标签去重"""
        return list(dict.fromkeys(v))

    @model_validator(mode='after')
    def check_due_date(self) -> 'TodoCreate':
        if self.due_date and self.due_date < datetime.now():
            raise ValueError('截止日期不能早于当前时间')
        return self


class TodoUpdate(BaseModel):
    """PATCH 更新，所有字段可选"""
    title: str | None = Field(None, min_length=1, max_length=200)
    description: str | None = None
    priority: int | None = Field(None, ge=1, le=5)
    completed: bool | None = None
    due_date: datetime | None = None


class TodoResponse(BaseModel):
    id: int
    title: str
    description: str | None
    priority: int
    completed: bool
    created_at: datetime
    due_date: datetime | None

    model_config = {'from_attributes': True}  # 支持 ORM 模型

2.3 response_model 与状态码

from fastapi import status

@app.post(
    '/todos',
    response_model=TodoResponse,
    status_code=status.HTTP_201_CREATED,
    summary='创建待办事项',
    tags=['todos'],
)
async def create_todo(todo: TodoCreate) -> TodoResponse:
    """创建一条新的待办事项。

    - **title**: 待办标题（必填）
    - **priority**: 优先级 1-5，默认为 1
    - **due_date**: 截止时间（可选）
    """
    # 此处模拟数据库操作
    return TodoResponse(
        id=1,
        title=todo.title,
        description=todo.description,
        priority=todo.priority,
        completed=False,
        created_at=datetime.now(),
        due_date=todo.due_date,
    )

---

3. 异常处理

3.1 HTTPException 与自定义异常处理器

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import logging

logger = logging.getLogger(__name__)


# 自定义异常类
class AppError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code


# 注册自定义异常处理器
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
    return JSONResponse(
        status_code=exc.status_code,
        content={'code': exc.code, 'message': exc.message},
    )


# 覆盖默认的 422 校验错误格式
@app.exception_handler(RequestValidationError)
async def validation_error_handler(
    request: Request,
    exc: RequestValidationError,
) -> JSONResponse:
    errors = []
    for error in exc.errors():
        errors.append({
            'field': '.'.join(str(loc) for loc in error['loc'][1:]),
            'message': error['msg'],
            'type': error['type'],
        })
    logger.warning('Validation error: %s %s -> %s', request.method, request.url, errors)
    return JSONResponse(
        status_code=422,
        content={'code': 'VALIDATION_ERROR', 'errors': errors},
    )


# 使用示例
@app.get('/todos/{todo_id}')
async def get_todo(todo_id: int) -> TodoResponse:
    # 模拟从数据库查询
    todo = None  # db.query(...)
    if todo is None:
        raise AppError('TODO_NOT_FOUND', f'待办 {todo_id} 不存在', status_code=404)
    return todo

---

4. 依赖注入

4.1 基础依赖

from fastapi import Depends
from typing import Annotated


def get_pagination(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
) -> dict:
    return {'offset': (page - 1) * size, 'limit': size}


Pagination = Annotated[dict, Depends(get_pagination)]


@app.get('/todos')
async def list_todos(pagination: Pagination) -> dict:
    return {'offset': pagination['offset'], 'limit': pagination['limit'], 'items': []}

4.2 数据库 Session 依赖（异步 SQLAlchemy 2.0）

# database.py
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker,
)
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = 'sqlite+aiosqlite:///./todos.db'

engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    pass


async def get_db() -> AsyncSession:
    """数据库会话依赖，确保用完后关闭"""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


DBSession = Annotated[AsyncSession, Depends(get_db)]

4.3 ORM 模型

# models.py
from sqlalchemy import String, Boolean, Integer, DateTime, Text, func
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime
from .database import Base


class Todo(Base):
    __tablename__ = 'todos'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    priority: Mapped[int] = mapped_column(Integer, default=1)
    completed: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), nullable=False
    )
    due_date: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    user_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)

---

5. JWT 认证

5.1 密码哈希与 Token 生成

# auth.py
from __future__ import annotations
import jwt
from datetime import datetime, timedelta, timezone
from pwdlib import PasswordHash
from pwdlib.hashers.argon2 import Argon2Hasher
from fastapi import HTTPException, status
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = 'your-256-bit-secret-change-in-production'
ALGORITHM = 'HS256'
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_hasher = PasswordHash([Argon2Hasher()])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth/token')


def hash_password(plain: str) -> str:
    return pwd_hasher.hash(plain)


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_hasher.verify(plain, hashed)


def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    to_encode.update({'exp': expire, 'iat': datetime.now(timezone.utc)})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='Token 已过期',
            headers={'WWW-Authenticate': 'Bearer'},
        )
    except jwt.PyJWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='Token 无效',
            headers={'WWW-Authenticate': 'Bearer'},
        )

5.2 认证依赖

# deps.py
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import Annotated
from .auth import oauth2_scheme, decode_token
from .models import User
from .database import get_db

DBSession = Annotated[AsyncSession, Depends(get_db)]


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: DBSession,
) -> User:
    payload = decode_token(token)
    user_id: int = payload.get('sub')
    if user_id is None:
        raise HTTPException(status_code=401, detail='Token 无效')
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if user is None:
        raise HTTPException(status_code=401, detail='用户不存在')
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]

5.3 登录端点

from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select

@app.post('/auth/token')
async def login(
    form: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: DBSession,
) -> dict:
    result = await db.execute(select(User).where(User.username == form.username))
    user = result.scalar_one_or_none()
    if not user or not verify_password(form.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='用户名或密码错误',
            headers={'WWW-Authenticate': 'Bearer'},
        )
    token = create_access_token({'sub': str(user.id), 'username': user.username})
    return {'access_token': token, 'token_type': 'bearer'}

---

6. 完整 CRUD 路由

# routers/todos.py
from fastapi import APIRouter, status
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated
from fastapi import Depends

from ..models import Todo
from ..schemas import TodoCreate, TodoUpdate, TodoResponse, PaginatedResponse
from ..deps import CurrentUser, DBSession

router = APIRouter(prefix='/todos', tags=['todos'])


@router.post('/', response_model=TodoResponse, status_code=201)
async def create_todo(
    todo_in: TodoCreate,
    current_user: CurrentUser,
    db: DBSession,
) -> Todo:
    todo = Todo(
        title=todo_in.title,
        description=todo_in.description,
        priority=todo_in.priority,
        due_date=todo_in.due_date,
        user_id=current_user.id,
    )
    db.add(todo)
    await db.flush()
    await db.refresh(todo)
    return todo


@router.get('/', response_model=PaginatedResponse[TodoResponse])
async def list_todos(
    current_user: CurrentUser,
    db: DBSession,
    page: int = 1,
    size: int = 20,
    completed: bool | None = None,
) -> dict:
    stmt = select(Todo).where(Todo.user_id == current_user.id)
    if completed is not None:
        stmt = stmt.where(Todo.completed == completed)

    count_stmt = select(func.count()).select_from(stmt.subquery())
    total = (await db.execute(count_stmt)).scalar()

    items_stmt = stmt.offset((page - 1) * size).limit(size).order_by(Todo.id.desc())
    items = (await db.execute(items_stmt)).scalars().all()

    return {'total': total, 'page': page, 'size': size, 'items': items}


@router.get('/{todo_id}', response_model=TodoResponse)
async def get_todo(
    todo_id: int,
    current_user: CurrentUser,
    db: DBSession,
) -> Todo:
    result = await db.execute(
        select(Todo).where(Todo.id == todo_id, Todo.user_id == current_user.id)
    )
    todo = result.scalar_one_or_none()
    if todo is None:
        raise HTTPException(status_code=404, detail='待办不存在')
    return todo


@router.patch('/{todo_id}', response_model=TodoResponse)
async def update_todo(
    todo_id: int,
    todo_in: TodoUpdate,
    current_user: CurrentUser,
    db: DBSession,
) -> Todo:
    result = await db.execute(
        select(Todo).where(Todo.id == todo_id, Todo.user_id == current_user.id)
    )
    todo = result.scalar_one_or_none()
    if todo is None:
        raise HTTPException(status_code=404, detail='待办不存在')

    update_data = todo_in.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(todo, field, value)

    await db.flush()
    await db.refresh(todo)
    return todo


@router.delete('/{todo_id}', status_code=204)
async def delete_todo(
    todo_id: int,
    current_user: CurrentUser,
    db: DBSession,
) -> None:
    result = await db.execute(
        select(Todo).where(Todo.id == todo_id, Todo.user_id == current_user.id)
    )
    todo = result.scalar_one_or_none()
    if todo is None:
        raise HTTPException(status_code=404, detail='待办不存在')
    await db.delete(todo)

---

7. 应用初始化

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .database import engine, Base
from .routers import todos, auth


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期：启动时建表，关闭时清理连接"""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()


app = FastAPI(
    title='Todo API',
    version='1.0.0',
    description='基于 FastAPI + SQLAlchemy 2.0 的 Todo 管理系统',
    lifespan=lifespan,
)

app.include_router(auth.router, tags=['auth'])
app.include_router(todos.router)


@app.get('/health')
async def health_check() -> dict:
    return {'status': 'ok'}

---

8. 使用 httpx 测试

# tests/test_todos.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.main import app
from app.database import get_db, Base


# 使用内存数据库测试
TEST_DB_URL = 'sqlite+aiosqlite:///:memory:'
test_engine = create_async_engine(TEST_DB_URL)
TestSessionLocal = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)


async def override_get_db():
    async with TestSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


app.dependency_overrides[get_db] = override_get_db


@pytest.fixture(autouse=True)
async def setup_db():
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url='http://test'
    ) as ac:
        yield ac


@pytest.fixture
async def auth_headers(client: AsyncClient) -> dict:
    """注册并登录，返回认证头"""
    await client.post('/auth/register', json={
        'username': 'testuser',
        'email': 'test@example.com',
        'password': 'TestPass123!',
    })
    resp = await client.post('/auth/token', data={
        'username': 'testuser',
        'password': 'TestPass123!',
    })
    token = resp.json()['access_token']
    return {'Authorization': f'Bearer {token}'}


@pytest.mark.asyncio
async def test_create_todo(client, auth_headers):
    resp = await client.post('/todos/', json={
        'title': '学习 FastAPI',
        'priority': 3,
    }, headers=auth_headers)
    assert resp.status_code == 201
    data = resp.json()
    assert data['title'] == '学习 FastAPI'
    assert data['priority'] == 3
    assert data['completed'] is False


@pytest.mark.asyncio
async def test_list_todos_pagination(client, auth_headers):
    for i in range(5):
        await client.post('/todos/', json={'title': f'Todo {i}'}, headers=auth_headers)

    resp = await client.get('/todos/?page=1&size=3', headers=auth_headers)
    assert resp.status_code == 200
    data = resp.json()
    assert data['total'] == 5
    assert len(data['items']) == 3


@pytest.mark.asyncio
async def test_update_todo(client, auth_headers):
    create_resp = await client.post('/todos/', json={'title': '原始标题'}, headers=auth_headers)
    todo_id = create_resp.json()['id']

    resp = await client.patch(f'/todos/{todo_id}', json={
        'title': '更新后标题',
        'completed': True,
    }, headers=auth_headers)
    assert resp.status_code == 200
    assert resp.json()['completed'] is True


@pytest.mark.asyncio
async def test_unauthorized_access(client):
    resp = await client.get('/todos/')
    assert resp.status_code == 401

运行测试：

pytest tests/ -v --asyncio-mode=auto

---

9. 部署

9.1 Uvicorn + Gunicorn

# 开发环境（热重载）
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 生产环境（多进程）
gunicorn app.main:app \
    -w 4 \
    -k uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 60 \
    --access-logfile -

9.2 Dockerfile

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
RUN adduser --disabled-password --no-create-home appuser
USER appuser

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

9.3 docker-compose.yml

version: '3.9'
services:
  api:
    build: .
    ports:
      - '8000:8000'
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:password@db:5432/todos
      SECRET_KEY: ${SECRET_KEY}
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: todos
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: password
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ['CMD-SHELL', 'pg_isready -U postgres']
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  pgdata:

---

10. 常见问题

问题	原因	解决方案
422 Validation Error 格式难看	默认错误格式	自定义 RequestValidationError 处理器
异步函数忘记 await	常见笔误	使用 asyncio 的类型检查工具
expire_on_commit=True 导致 lazy load 错误	Session 关闭后访问属性	async_sessionmaker 设置 expire_on_commit=False
Pydantic v2 orm_mode 报错	v2 已改名	使用 model_config = {'from_attributes': True}
依赖注入的 yield 不执行	异常未被捕获	在 get_db 中加 try/except/finally
CORS 跨域问题	前后端分离	app.add_middleware(CORSMiddleware, ...)

---

11. 小结

知识点	核心要点
路径/查询参数	Path()、Query() 注解校验
Pydantic v2	@field_validator、@model_validator、model_dump(exclude_unset=True)
依赖注入	Depends() + yield + Annotated
异步数据库	SQLAlchemy 2.0 + async_sessionmaker + Mapped 类型注解
JWT 认证	pyjwt + pwdlib(Argon2) + OAuth2PasswordBearer
测试	httpx AsyncClient + ASGITransport + 依赖覆盖
部署	Uvicorn/Gunicorn + Docker + docker-compose

---

上一篇【第026篇】CLI 工具开发——Click + Rich 打造专业命令行应用
下一篇【第028篇】自动化脚本实战——文件处理、定时任务与 Web 爬虫

---

参考资料

1. FastAPI 官方文档
2. Pydantic v2 官方文档
3. SQLAlchemy 2.0 异步文档
4. PyJWT 文档
5. httpx 测试文档