> ## Content Index
> Fetch the complete content index at: https://blog.vercanti.com/llms.txt
> Use this file to discover other available public pages before exploring further.

# FastAPI 完全指南
- URL: https://blog.vercanti.com/fastapi-wan-quan-zhi-nan/
- Published: 2026-08-28T14:34:40.000Z
- Updated: 2026-08-28T14:57:04.000Z
- Description: FastAPI 是一个用于构建 API 的现代、高性能 Python Web 框架，基于标准 Python 类型提示。主要特点： ASGI（Asynchronous Server Gateway Interface）是 WSGI 的异步升级版。FastAPI 基于 Starlette，Starlette 实现了 ASGI 接口。运行 FastAPI 需要 ASGI 服务器，如 uvicorn 或 hypercorn。 FastAPI 自动根据路由和 Pydantic 模型生成两套文档： Path() 参数表： Query() 参数表： 注意：Pydant
- Author: yellowdog
- Tags: Python, 框架与库

> 官方文档：<https://fastapi.tiangolo.com/zh/>  
> 最后更新：2026-03-05

---

## 1\. 基础概念

### FastAPI 是什么

FastAPI 是一个用于构建 API 的现代、高性能 Python Web 框架，基于标准 Python 类型提示。主要特点：

- 基于 ASGI（Starlette），支持异步
- 自动生成交互式 API 文档（Swagger UI / ReDoc）
- 利用 Pydantic 进行数据验证和序列化
- 性能接近 NodeJS 和 Go（基于 Starlette 和 Pydantic）

### 与 Flask / Django 的对比

| 特性     | FastAPI     | Flask  | Django        |
| ------ | ----------- | ------ | ------------- |
| 接口规范   | ASGI        | WSGI   | WSGI          |
| 异步支持   | 原生支持        | 需要扩展   | 部分支持          |
| 数据验证   | Pydantic 内置 | 需要第三方库 | 表单/序列化器       |
| API 文档 | 自动生成        | 需要扩展   | 需要 DRF        |
| 学习曲线   | 中等          | 低      | 高             |
| 适合场景   | API 服务、微服务  | 小型应用   | 全栈 Web 应用     |
| ORM    | 无内置         | 无内置    | Django ORM 内置 |

### ASGI

ASGI（Asynchronous Server Gateway Interface）是 WSGI 的异步升级版。FastAPI 基于 Starlette，Starlette 实现了 ASGI 接口。运行 FastAPI 需要 ASGI 服务器，如 uvicorn 或 hypercorn。

### 自动文档

FastAPI 自动根据路由和 Pydantic 模型生成两套文档：

- **Swagger UI**：访问 `/docs`，支持在线测试接口
- **ReDoc**：访问 `/redoc`，适合阅读文档
- **OpenAPI JSON**：访问 `/openapi.json`，原始 schema

---

## 2\. 快速开始

### 安装

```bash
# 安装 FastAPI 和 ASGI 服务器
pip install fastapi uvicorn[standard]

# 可选：安装 python-multipart（文件上传需要）
pip install python-multipart

# 可选：安装 python-jose（JWT 认证需要）
pip install python-jose[cryptography] passlib[bcrypt]

```

### 第一个应用

```python
from fastapi import FastAPI

app = FastAPI(
    title="示例 API",          # 文档标题
    description="这是一个示例",  # 文档描述
    version="1.0.0",           # API 版本
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

```

### 运行（uvicorn 参数表）

```bash
uvicorn main:app --reload

```

| 参数               | 类型   | 默认值       | 说明                                     |
| ---------------- | ---- | --------- | -------------------------------------- |
| app              | str  | 必填        | 格式为 模块名:实例名                            |
| \--host          | str  | 127.0.0.1 | 监听地址，0.0.0.0 对外开放                      |
| \--port          | int  | 8000      | 监听端口                                   |
| \--reload        | flag | False     | 代码变更时自动重启，仅开发环境使用                      |
| \--workers       | int  | 1         | 工作进程数，生产环境建议 CPU 核心数 \* 2 + 1          |
| \--log-level     | str  | info      | 日志级别：debug/info/warning/error/critical |
| \--ssl-keyfile   | str  | None      | SSL 私钥文件路径                             |
| \--ssl-certfile  | str  | None      | SSL 证书文件路径                             |
| \--proxy-headers | flag | False     | 信任反向代理传来的 X-Forwarded-\* 头             |
| \--root-path     | str  | ""        | 设置 ASGI root\_path，部署在子路径时使用           |

---

## 3\. 路径操作

### 路径参数

```python
from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(
        ...,               # 必填（... 表示必填）
        title="商品ID",
        description="商品的唯一标识符",
        ge=1,              # 大于等于 1
        le=1000,           # 小于等于 1000
    )
):
    return {"item_id": item_id}

```

**Path() 参数表：**

| 参数                  | 类型    | 默认值   | 说明                                                                                                |
| ------------------- | ----- | ----- | ------------------------------------------------------------------------------------------------- |
| default             | Any   | 必填    | 默认值，... 表示必填                                                                                      |
| title               | str   | None  | 文档中显示的标题                                                                                          |
| description         | str   | None  | 文档中显示的描述                                                                                          |
| alias               | str   | None  | 参数别名（客户端传参时使用）                                                                                    |
| gt                  | float | None  | 大于（greater than）                                                                                  |
| ge                  | float | None  | 大于等于（greater than or equal）                                                                       |
| lt                  | float | None  | 小于（less than）                                                                                     |
| le                  | float | None  | 小于等于（less than or equal）                                                                          |
| min\_length         | int   | None  | 字符串最小长度                                                                                           |
| max\_length         | int   | None  | 字符串最大长度                                                                                           |
| pattern             | str   | None  | 字符串正则表达式校验，见 [正则表达式完全指南](https://blog.vercanti.com/python-zheng-ze-biao-da-shi-wan-quan-zhi-nan/) |
| deprecated          | bool  | False | 标记为已废弃                                                                                            |
| include\_in\_schema | bool  | True  | 是否显示在文档中                                                                                          |

### 查询参数

```python
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(
    q: str = Query(None, min_length=3, max_length=50),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, le=100),
    tags: list[str] = Query(default=[]),  # 支持多值：?tags=a&tags=b
):
    return {"q": q, "skip": skip, "limit": limit, "tags": tags}

```

**Query() 参数表：**

| 参数                        | 类型    | 默认值   | 说明           |
| ------------------------- | ----- | ----- | ------------ |
| default                   | Any   | 必填    | 默认值，... 表示必填 |
| title                     | str   | None  | 文档标题         |
| description               | str   | None  | 文档描述         |
| alias                     | str   | None  | 参数别名         |
| gt / ge / lt / le         | float | None  | 数值范围校验       |
| min\_length / max\_length | int   | None  | 字符串长度校验      |
| pattern                   | str   | None  | 正则表达式校验      |
| deprecated                | bool  | False | 标记已废弃        |
| include\_in\_schema       | bool  | True  | 是否显示在文档中     |

### 请求体（Pydantic BaseModel）

```python
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/")
async def create_item(item: Item):
    # item 是经过验证的 Item 实例
    item_dict = item.model_dump()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

```

---

## 4\. Pydantic 模型

### Field() 参数表

```python
from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(
        ...,
        title="商品名称",
        description="商品的显示名称",
        min_length=1,
        max_length=100,
        examples=["笔记本电脑"],
    )
    price: float = Field(..., gt=0, description="价格，必须大于0")
    quantity: int = Field(default=0, ge=0)

```

| 参数                        | 类型       | 默认值   | 说明                |
| ------------------------- | -------- | ----- | ----------------- |
| default                   | Any      | 必填    | 默认值，... 表示必填      |
| default\_factory          | callable | None  | 动态默认值工厂函数，如 list  |
| title                     | str      | None  | 字段标题              |
| description               | str      | None  | 字段描述              |
| alias                     | str      | None  | 字段别名（序列化/反序列化时使用） |
| gt / ge / lt / le         | float    | None  | 数值范围              |
| min\_length / max\_length | int      | None  | 字符串长度             |
| pattern                   | str      | None  | 正则表达式             |
| examples                  | list     | None  | 示例值列表             |
| exclude                   | bool     | False | 序列化时排除该字段         |
| frozen                    | bool     | False | 是否禁止修改（只读字段）      |
| validate\_default         | bool     | False | 是否对默认值也进行验证       |

### 类型验证与可选字段

```python
from typing import Optional
from pydantic import BaseModel

class UserProfile(BaseModel):
    username: str                           # 必填
    email: str                              # 必填
    age: Optional[int] = None               # 可选，默认 None（Python 3.9 以下写法）
    bio: str | None = None                  # 可选，默认 None（Python 3.10+ 写法）
    tags: list[str] = []                    # 可选，默认空列表
    metadata: dict[str, str] = {}           # 可选，默认空字典

```

### 嵌套模型

```python
from pydantic import BaseModel

class Address(BaseModel):
    street: str
    city: str
    country: str = "China"

class User(BaseModel):
    name: str
    address: Address                        # 嵌套模型
    addresses: list[Address] = []           # 嵌套模型列表

```

### validator / @field\_validator（Pydantic v2）

```python
from pydantic import BaseModel, field_validator, model_validator

class UserCreate(BaseModel):
    username: str
    password: str
    confirm_password: str
    email: str

    # 单字段验证（Pydantic v2 写法）
    @field_validator("username")
    @classmethod
    def username_must_be_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError("用户名只能包含字母和数字")
        return v.lower()  # 可以对值进行转换

    @field_validator("email")
    @classmethod
    def email_must_contain_at(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("邮箱格式不正确")
        return v

    # 多字段交叉验证（Pydantic v2 写法）
    @model_validator(mode="after")
    def check_passwords_match(self) -> "UserCreate":
        if self.password != self.confirm_password:
            raise ValueError("两次密码不一致")
        return self

```

**注意**：Pydantic v1 使用 `@validator`，v2 使用 `@field_validator`，两者语法不同，详见第 17 节。

---

## 5\. 响应模型

### response\_model

```python
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    username: str
    password: str          # 输入模型包含密码

class UserPublic(BaseModel):
    id: int
    username: str          # 输出模型不暴露密码

@app.post("/users/", response_model=UserPublic)
async def create_user(user: UserCreate):
    # 返回的数据会按 UserPublic 过滤，password 不会出现在响应中
    return {"id": 1, "username": user.username, "password": user.password}

```

### response\_model\_exclude\_unset

```python
@app.patch("/users/{user_id}", response_model=UserPublic, response_model_exclude_unset=True)
async def update_user(user_id: int, user: UserCreate):
    # 只返回实际设置的字段，未设置的字段不出现在响应中
    return user

```

### 状态码

```python
from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    return None  # 204 不返回响应体

```

常用状态码常量（`fastapi.status` 模块）：

| 常量                                 | 数值  | 含义               |
| ---------------------------------- | --- | ---------------- |
| HTTP\_200\_OK                      | 200 | 成功               |
| HTTP\_201\_CREATED                 | 201 | 创建成功             |
| HTTP\_204\_NO\_CONTENT             | 204 | 无内容              |
| HTTP\_400\_BAD\_REQUEST            | 400 | 请求错误             |
| HTTP\_401\_UNAUTHORIZED            | 401 | 未认证              |
| HTTP\_403\_FORBIDDEN               | 403 | 无权限              |
| HTTP\_404\_NOT\_FOUND              | 404 | 资源不存在            |
| HTTP\_422\_UNPROCESSABLE\_ENTITY   | 422 | 验证失败（FastAPI 默认） |
| HTTP\_500\_INTERNAL\_SERVER\_ERROR | 500 | 服务器内部错误          |

---

## 6\. 依赖注入

### Depends() 基础用法

```python
from fastapi import FastAPI, Depends

app = FastAPI()

# 定义依赖函数
async def get_current_user(token: str = Query(...)):
    # 从 token 中解析用户
    user = verify_token(token)
    return user

@app.get("/users/me")
async def read_users_me(current_user = Depends(get_current_user)):
    return current_user

```

### 类依赖

```python
class CommonQueryParams:
    def __init__(self, skip: int = 0, limit: int = 10, q: str | None = None):
        self.skip = skip
        self.limit = limit
        self.q = q

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    # 可简写为 Depends()，FastAPI 自动推断
    return {"skip": commons.skip, "limit": commons.limit, "q": commons.q}

```

### 多层依赖

```python
async def verify_token(token: str = Header(...)):
    if token != "secret":
        raise HTTPException(status_code=401, detail="无效 token")
    return token

async def verify_key(api_key: str = Header(...)):
    if api_key != "my-api-key":
        raise HTTPException(status_code=403, detail="无效 API Key")
    return api_key

@app.get("/protected/")
async def protected_route(
    token = Depends(verify_token),
    key = Depends(verify_key),
):
    return {"token": token, "key": key}

```

### 数据库会话依赖（与 SQLAlchemy 集成）

```python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session  # yield 使依赖支持上下文管理器

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    # 请求结束后 session 自动关闭
    user = await db.get(User, user_id)
    return user

```

详见 [SQLAlchemy完全指南](https://blog.vercanti.com/sqlalchemy-wan-quan-zhi-nan/)。

---

## 7\. 路由

### APIRouter

```python
# routers/items.py
from fastapi import APIRouter

router = APIRouter(
    prefix="/items",        # 所有路由加前缀 /items
    tags=["items"],         # 文档中的分组标签
    dependencies=[Depends(verify_token)],  # 该路由下所有接口共用的依赖
    responses={404: {"description": "Not found"}},  # 通用响应说明
)

@router.get("/")
async def list_items():
    return []

@router.get("/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id}

```

**APIRouter() 参数表：**

| 参数                       | 类型          | 默认值          | 说明                 |
| ------------------------ | ----------- | ------------ | ------------------ |
| prefix                   | str         | ""           | 路由前缀，必须以 / 开头或为空   |
| tags                     | list\[str\] | None         | 文档分组标签             |
| dependencies             | list        | None         | 该路由下所有接口共享的依赖      |
| responses                | dict        | None         | 通用响应定义（出现在每个接口文档中） |
| deprecated               | bool        | False        | 标记整个路由为已废弃         |
| include\_in\_schema      | bool        | True         | 是否在文档中显示           |
| default\_response\_class | type        | JSONResponse | 默认响应类              |

### 路由注册

```python
# main.py
from fastapi import FastAPI
from routers import items, users

app = FastAPI()

app.include_router(items.router)
app.include_router(users.router, prefix="/api/v1")  # 可覆盖前缀

```

---

## 8\. 中间件

### @app.middleware("http")

```python
import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)  # 调用下一个处理器
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

```

### CORSMiddleware

```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

```

**CORSMiddleware 参数表：**

| 参数                   | 类型          | 默认值       | 说明                                                |
| -------------------- | ----------- | --------- | ------------------------------------------------- |
| allow\_origins       | list\[str\] | \[\]      | 允许的来源列表，\["\*"\] 表示允许所有                           |
| allow\_origin\_regex | str         | None      | 允许来源的正则表达式                                        |
| allow\_methods       | list\[str\] | \["GET"\] | 允许的 HTTP 方法，\["\*"\] 表示所有方法                       |
| allow\_headers       | list\[str\] | \[\]      | 允许的请求头，\["\*"\] 表示所有头                             |
| allow\_credentials   | bool        | False     | 是否允许携带 Cookie/认证信息（为 True 时 origins 不能为 \["\*"\]） |
| expose\_headers      | list\[str\] | \[\]      | 允许浏览器访问的响应头                                       |
| max\_age             | int         | 600       | 预检请求的缓存时间（秒）                                      |

### 自定义中间件（基于类）

```python
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        token = request.headers.get("Authorization")
        if not token:
            from starlette.responses import JSONResponse
            return JSONResponse({"detail": "未授权"}, status_code=401)
        response = await call_next(request)
        return response

app.add_middleware(AuthMiddleware)

```

---

## 9\. 认证

### OAuth2PasswordBearer

```python
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()

# tokenUrl 是获取 token 的接口路径
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

@app.post("/auth/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # form_data.username, form_data.password
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="用户名或密码错误")
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

```

### JWT Token（python-jose）

```python
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = "your-secret-key"  # 生产环境使用环境变量
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

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=15))
    to_encode.update({"exp": expire})
    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 JWTError:
        raise HTTPException(status_code=401, detail="Token 无效或已过期")

```

### 依赖注入保护路由

```python
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = decode_token(token)
    username = payload.get("sub")
    if not username:
        raise HTTPException(status_code=401, detail="无效凭证")
    user = await get_user_by_username(username)
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

async def get_current_active_user(current_user = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="用户已禁用")
    return current_user

@app.get("/users/me")
async def read_users_me(current_user = Depends(get_current_active_user)):
    return current_user

```

---

## 10\. 文件上传

### 单文件上传

```python
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()          # 读取文件内容（bytes）
    return {
        "filename": file.filename,        # 原始文件名
        "content_type": file.content_type, # MIME 类型
        "size": len(contents),
    }

```

### UploadFile 属性和方法

| 属性/方法         | 类型        | 说明                       |
| ------------- | --------- | ------------------------ |
| filename      | str       | 原始文件名                    |
| content\_type | str       | MIME 类型，如 image/jpeg     |
| headers       | Headers   | 文件头信息                    |
| size          | int       | 文件大小（字节），Python 3.11+ 可用 |
| read(size)    | coroutine | 读取文件内容，返回 bytes          |
| write(data)   | coroutine | 写入内容                     |
| seek(offset)  | coroutine | 移动文件指针                   |
| close()       | coroutine | 关闭文件                     |

### 多文件上传

```python
@app.post("/upload/multiple/")
async def upload_multiple_files(files: list[UploadFile] = File(...)):
    results = []
    for file in files:
        contents = await file.read()
        results.append({"filename": file.filename, "size": len(contents)})
    return results

```

### 保存文件到磁盘

```python
import aiofiles

@app.post("/upload/save/")
async def upload_and_save(file: UploadFile = File(...)):
    save_path = f"/tmp/{file.filename}"
    async with aiofiles.open(save_path, "wb") as f:
        while chunk := await file.read(1024 * 1024):  # 每次读取 1MB
            await f.write(chunk)
    return {"saved_to": save_path}

```

---

## 11\. WebSocket

### WebSocket 连接

```python
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()             # 接受连接
    try:
        while True:
            data = await websocket.receive_text()   # 接收文本
            await websocket.send_text(f"收到: {data}")  # 发送文本
    except Exception:
        await websocket.close()

```

### 广播示例（多客户端）

```python
from fastapi import FastAPI, WebSocket
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"客户端 {client_id} 说: {data}")
    except Exception:
        manager.disconnect(websocket)
        await manager.broadcast(f"客户端 {client_id} 已离线")

```

---

## 12\. 后台任务

### BackgroundTasks.add\_task()

```python
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def send_email_notification(email: str, message: str):
    # 模拟耗时操作，在后台线程执行（同步函数）
    import time
    time.sleep(5)
    print(f"发送邮件到 {email}: {message}")

async def log_operation(operation: str):
    # 异步后台任务
    await some_db_operation(operation)

@app.post("/send-notification/")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks,
):
    # add_task 的参数：func, *args, **kwargs
    background_tasks.add_task(send_email_notification, email, "操作成功")
    background_tasks.add_task(log_operation, "send_notification")
    return {"message": "通知已加入队列"}

```

**注意**：BackgroundTasks 在响应返回后在同一进程中执行，适合轻量任务。重量级任务请使用 Celery 或 ARQ。

---

## 13\. 异常处理

### HTTPException

```python
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(
            status_code=404,
            detail="商品不存在",         # 可以是字符串或字典
            headers={"X-Error": "true"},  # 可选：附加响应头
        )
    return items_db[item_id]

```

### 自定义异常处理器

```python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class BusinessException(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message

@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, exc: BusinessException):
    return JSONResponse(
        status_code=400,
        content={"code": exc.code, "message": exc.message},
    )

@app.exception_handler(404)
async def not_found_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=404,
        content={"message": "资源不存在"},
    )

```

### 覆盖默认验证错误处理

```python
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body},
    )

```

---

## 14\. 测试

### TestClient（httpx）

```python
# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "笔记本", "price": 99.9},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "笔记本"

```

### pytest fixture（依赖注入覆盖）

```python
import pytest
from fastapi.testclient import TestClient
from main import app
from dependencies import get_db

# 测试用数据库
def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

@pytest.fixture
def client():
    # 覆盖依赖项
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def test_create_user(client):
    response = client.post("/users/", json={"username": "test", "password": "secret"})
    assert response.status_code == 201

```

### 异步测试

```python
import pytest
import httpx
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200

```

---

## 15\. 部署

### uvicorn + gunicorn

```bash
# 安装
pip install gunicorn uvicorn[standard]

# 运行（gunicorn 管理进程，uvicorn 作为 worker）
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

```

### Docker

```dockerfile
FROM python:3.12-slim

WORKDIR /app

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

COPY . .

# 生产环境运行
CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

```

### 环境变量配置（pydantic-settings）

```python
# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///./test.db"
    secret_key: str = "dev-secret-key"
    debug: bool = False
    allowed_origins: list[str] = ["http://localhost:3000"]

    class Config:
        env_file = ".env"           # 从 .env 文件读取
        env_file_encoding = "utf-8"

settings = Settings()

# 使用
@app.get("/info")
async def info():
    return {"debug": settings.debug}

```

`.env` 文件：

```
DATABASE_URL=postgresql+asyncpg://user:password@localhost/mydb
SECRET_KEY=production-secret-key-change-this
DEBUG=false

```

---

## 16\. 最佳实践

### 项目目录结构

```
myproject/
├── main.py                 # FastAPI 应用入口
├── config.py               # 配置管理（pydantic-settings）
├── database.py             # 数据库连接和 session
├── models/                 # SQLAlchemy ORM 模型
│   ├── __init__.py
│   ├── user.py
│   └── item.py
├── schemas/                # Pydantic 输入输出模型
│   ├── __init__.py
│   ├── user.py
│   └── item.py
├── routers/                # APIRouter 路由模块
│   ├── __init__.py
│   ├── users.py
│   └── items.py
├── dependencies/           # 可复用依赖项
│   ├── __init__.py
│   └── auth.py
├── services/               # 业务逻辑层（可选）
│   ├── __init__.py
│   └── user_service.py
└── tests/
    ├── conftest.py
    └── test_users.py

```

### 异步数据库（与 SQLAlchemy 集成）

详见 [SQLAlchemy完全指南](https://blog.vercanti.com/sqlalchemy-wan-quan-zhi-nan/)。

```python
# database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,          # 连接池大小
    max_overflow=20,       # 超出 pool_size 时最多再创建的连接数
    echo=False,            # 生产环境关闭 SQL 日志
)

AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

```

### lifespan（应用生命周期，推荐写法）

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 应用启动时执行
    await create_db_tables()
    print("应用已启动")
    yield
    # 应用关闭时执行
    await engine.dispose()
    print("应用已关闭")

app = FastAPI(lifespan=lifespan)

```

**注意**：`lifespan` 是推荐写法，替代已废弃的 `@app.on_event("startup")` 和 `@app.on_event("shutdown")`。

---

## 17\. 常见陷阱

### 同步函数 vs async

```python
# 陷阱：在 async 路由中调用同步阻塞函数会阻塞事件循环
@app.get("/bad/")
async def bad_endpoint():
    import time
    time.sleep(5)  # 错误！阻塞整个事件循环
    return {}

# 正确做法 1：使用 asyncio.sleep
import asyncio
@app.get("/good-sleep/")
async def good_sleep():
    await asyncio.sleep(5)  # 正确，不阻塞事件循环
    return {}

# 正确做法 2：阻塞操作放在线程池
import asyncio
@app.get("/good-blocking/")
async def good_blocking():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking_function)
    return {"result": result}

# 正确做法 3：路由函数定义为普通函数（FastAPI 自动放入线程池）
@app.get("/sync-route/")
def sync_route():
    import time
    time.sleep(5)  # 可以，FastAPI 会在线程池中运行
    return {}

```

### Pydantic v1 vs v2 区别

| 特性     | Pydantic v1         | Pydantic v2                         |
| ------ | ------------------- | ----------------------------------- |
| 验证器装饰器 | @validator("field") | @field\_validator("field")          |
| 模型方法   | .dict()             | .model\_dump()                      |
| 模型方法   | .json()             | .model\_dump\_json()                |
| 模型方法   | .copy()             | .model\_copy()                      |
| 模型方法   | .parse\_obj()       | .model\_validate()                  |
| 模型方法   | .schema()           | .model\_json\_schema()              |
| 验证器参数  | values              | 使用 @model\_validator(mode="before") |
| 配置类    | class Config        | model\_config = ConfigDict(...)     |
| 性能     | 基准性能                | v1 的 5-50x 倍                        |

FastAPI 0.100.0+ 默认使用 Pydantic v2，但保留了 v1 兼容层。

### CORS 常见错误

```python
# 错误：allow_credentials=True 时不能使用通配符 origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],          # 错误！
    allow_credentials=True,
)

# 正确：明确指定来源
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://frontend.example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

```

### 路由顺序问题

```python
# 陷阱：固定路径必须在参数路径之前定义
@app.get("/items/search")   # 如果放在 /{item_id} 后面，会被 /{item_id} 匹配到
async def search_items(): ...

@app.get("/items/{item_id}")
async def get_item(item_id: str): ...

```

### Response 对象泄漏

```python
# 陷阱：直接返回 ORM 对象（lazy 属性在 session 关闭后访问失败）
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)
    return user  # 如果 User 有懒加载关系，此处可能报错

# 正确：使用 response_model 或先序列化
@app.get("/users/{user_id}", response_model=UserPublic)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)
    return UserPublic.model_validate(user)  # 在 session 关闭前完成序列化

```

---

## 最佳实践

**依赖注入用 `Annotated` 类型封装**：将 `Depends(get_db)` 包在 `Annotated` 里，路由签名更简洁，且可全局复用：

```python
from typing import Annotated
DbDep = Annotated[AsyncSession, Depends(get_db)]

@router.get("/users")
async def list_users(db: DbDep):
    ...

```

**用 `APIRouter` 按业务模块拆分路由**：不要把所有路由都写在 `main.py`，每个业务域一个 router 文件，在 `main.py` 用 `app.include_router` 聚合，prefix 和 tags 在 include 时统一设置。

**`response_model` 始终显式声明**：FastAPI 依赖 `response_model` 过滤和验证响应，未声明时直接序列化 ORM 对象可能暴露敏感字段（如密码哈希）。

**背景任务用 `BackgroundTasks`，长任务用 Celery**：`BackgroundTasks` 在响应发送后同步运行，适合发邮件等快速任务；CPU 密集或需要重试的任务放 Celery worker，避免阻塞 ASGI worker。

**中间件顺序影响请求链**：`app.add_middleware` 后添加的中间件先执行（洋葱模型），CORS 中间件必须最外层（最后 add），否则预检请求会被其他中间件拦截返回非 200。

---

## 常见陷阱

### 陷阱：在路由函数中直接修改 Pydantic 模型实例

**现象：** 修改从请求体解析的 schema 对象后，后续验证或序列化行为异常。  
**原因：** Pydantic v2 默认 model 是不可变的，直接修改字段会静默失败或抛出验证错误。  
**解决：** 用 `model.model_copy(update={...})` 创建修改后的新实例。

### 陷阱：`async def` 路由中调用同步阻塞函数

**现象：** 高并发时 FastAPI 响应明显变慢，`/metrics` 显示 active requests 堆积。  
**原因：** `async def` 路由运行在事件循环中，同步阻塞调用（`time.sleep`、同步 DB 查询）阻塞整个循环。  
**解决：** 同步阻塞代码用 `run_in_executor` 放线程池，或改用异步库（`httpx`、`asyncpg`）。

### 陷阱：生命周期事件用 `@app.on_event` 已废弃

**现象：** 代码中使用 `@app.on_event("startup")` 后，新版 FastAPI 打印 DeprecationWarning。  
**原因：** FastAPI 0.93+ 推荐用 `lifespan` context manager 替代 `on_event`，`on_event` 将在未来版本移除。  
**解决：** 改用 `@asynccontextmanager` \+ `lifespan` 参数：

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield
    await close_db()

app = FastAPI(lifespan=lifespan)

```

---

## 参见

- [SQLAlchemy完全指南](https://blog.vercanti.com/sqlalchemy-wan-quan-zhi-nan/)
- [Pydantic完全指南](https://blog.vercanti.com/pydantic-wan-quan-zhi-nan/)
- [asyncio异步编程完全指南](https://blog.vercanti.com/asyncio-yi-bu-bian-cheng-wan-quan-zhi-nan/)
- [pytest完全指南](https://blog.vercanti.com/pytest-wan-quan-zhi-nan/)
- [JWT完全指南](https://blog.vercanti.com/jwt-wan-quan-zhi-nan/)