一、概览与提问（SQ3R · Survey & Question）

SQ3R 第一步：快速浏览全貌，提出关键问题。

什么是 FastAPI？

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 Web 框架，使用 Python 并基于标准的 Python 类型提示（Type Hints）。它由 Sebastián Ramírez（@tiangolo）创建，自发布以来被 Microsoft、Uber、Netflix 等公司广泛采用。

FastAPI 的核心价值：

  • 快速：极高性能，可与 NodeJS 和 Go 并肩（归功于 Starlette 和 Pydantic），是最快的 Python Web 框架之一
  • 高效编码：功能开发速度提升约 200%～300%，参数声明一次即可获得数据校验、序列化、文档生成等多种功能
  • 更少 bug：人为（开发者）错误减少约 40%，类型系统在开发阶段就能捕获大量错误
  • 直觉化：极佳的编辑器支持，处处皆可自动补全
  • 标准化：基于并完全兼容 OpenAPI（原 Swagger）和 JSON Schema 标准

FastAPI 站在两个巨人的肩膀上：Starlette 负责 Web 层（路由、中间件、ASGI），Pydantic 负责数据层（校验、序列化、设置管理）。

核心问题

  • FastAPI 适合什么场景？——RESTful API、微服务、机器学习服务后端、异步 IO 密集型应用
  • FastAPI 与 Flask/Django REST 相比有什么优势？——原生异步、自动文档、类型驱动、更高性能
  • 学习 FastAPI 需要什么基础？——Python 3.10+、基本的 HTTP 知识、类型提示（Type Hints）
  • FastAPI 为什么这么快？——基于 ASGI 异步协议、Pydantic V2 Rust 内核、Starlette 高性能路由

技术全景图

FastAPI 应用
├── 核心基础
│   ├── 路径操作（Routing）——HTTP 方法与路径声明
│   ├── 路径参数（Path Parameters）——URL 中的动态变量
│   ├── 查询参数（Query Parameters）——URL ?key=value
│   ├── 请求体（Request Body）——Pydantic 模型定义 JSON 数据
│   ├── 响应模型（Response Model）——类型安全的返回值
│   └── 状态码（Status Codes）——HTTP 状态码声明
├── 进阶用法
│   ├── 依赖注入（Dependency Injection）——强大的 DI 系统
│   ├── 中间件（Middleware）——请求/响应拦截
│   ├── CORS ——跨域资源共享配置
│   ├── 安全与认证 ——OAuth2、JWT、HTTP Basic
│   ├── 数据库集成 ——SQLAlchemy 等 ORM 集成
│   ├── 后台任务（Background Tasks）
│   ├── WebSocket ——双向实时通信
│   └── 文件上传 & 表单数据
├── 深度原理
│   ├── ASGI 协议与 Starlette
│   ├── Pydantic V2 验证引擎（Rust 内核）
│   ├── OpenAPI 自动生成
│   ├── 测试策略（TestClient）
│   └── 部署方案（Docker、Uvicorn、Gunicorn）
└── 生态工具
    ├── FastAPI CLI ——开发与部署命令行工具
    ├── Swagger UI / ReDoc ——交互式 API 文档
    └── pydantic-settings ——配置管理

二、用最简单的话说清楚（费曼学习法）

费曼学习法核心理念：如果你不能用简单的语言解释一件事，说明你还没有真正理解它。

核心概念讲解

路径操作（Path Operation）

FastAPI 用"装饰器 + 函数"的方式定义 API 端点。@app.get("/") 意味着"当有人用 GET 方法访问 / 路径时，执行下面的函数"。

from fastapi import FastAPI
 
app = FastAPI()
 
@app.get("/")
def read_root():
    return {"Hello": "World"}

就这么简单——一个装饰器、一个函数、一个返回值。FastAPI 会自动把返回的字典转换为 JSON 响应。

路径参数（Path Parameters）

URL 中的动态部分用花括号 {} 声明，FastAPI 会自动提取并转换类型：

@app.get("/items/{item_id}")
def read_item(item_id: int):
    # item_id 自动转为 int，如果传入非数字则返回清晰的 422 错误
    return {"item_id": item_id}

注意 item_id: int ——这是标准的 Python 类型注解。FastAPI 利用它做数据校验和文档生成。

查询参数（Query Parameters）

函数参数不在路径中时，FastAPI 自动将其识别为查询参数：

@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
    # 访问 /items/?skip=5&limit=20
    return {"skip": skip, "limit": limit}

有默认值的参数是可选的，没有默认值的是必需的。

请求体与 Pydantic 模型（Request Body）

发送 JSON 数据到 API 时，用 Pydantic 的 BaseModel 声明数据结构：

from pydantic import BaseModel
 
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool | None = None
 
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

一次声明，你获得：自动 JSON 解析、数据校验（name 必须是字符串、price 必须是数字）、编辑器自动补全（item.name 有类型提示）、自动生成的 API 文档。

依赖注入（Dependency Injection）

依赖注入就是"让 FastAPI 帮你调用某些函数，把结果传给你的路由函数"。它是 FastAPI 最强大的特性之一：

from typing import Annotated
from fastapi import Depends, FastAPI
 
app = FastAPI()
 
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}
 
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
 
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

common_parameters 被两个路由共用——FastAPI 自动调用它并把结果注入。不需要注册、不需要继承、不需要装饰器以外的任何配置。

自动 API 文档

FastAPI 根据你的代码自动生成交互式 API 文档：

  • Swagger UI：访问 http://127.0.0.1:8000/docs——可以直接在浏览器中测试 API
  • ReDoc：访问 http://127.0.0.1:8000/redoc——美观的参考文档

你不需要写一行额外的文档代码。你写的类型注解就是文档。

类比与比喻

  • FastAPI 像一家全自动餐厅：你只需要写下菜单（类型注解），厨房会自动检查食材（校验）、烹饪（处理）、装盘（序列化）、打印菜单给客人（文档生成）
  • Pydantic 模型像海关检查：所有进入的数据都要过一道严格的检查——类型不对？拒绝进入并告诉你具体哪里有问题。合格的数据才被放行到你的业务逻辑
  • 依赖注入像外卖配送：你的路由函数不需要自己去取食材（数据库连接、用户认证信息），FastAPI 像"配送系统"一样，按需把结果送到你手中
  • ASGI 像"高速公路"vs WSGI 的"普通公路"：WSGI（Flask/Django 用的协议）一次只能处理一辆车，ASGI 可以同时处理多辆车，所以 FastAPI 能实现高并发

常见误解澄清

  1. 误解：FastAPI 需要写很多配置代码 —— 事实：FastAPI 几乎零配置，类型注解就是配置
  2. 误解：必须用 async def 才能获得高性能 —— 事实：FastAPI 会智能处理 def 和 async def，同步函数会在线程池中执行，不会阻塞事件循环
  3. 误解：FastAPI 只能构建 API —— 事实：通过 Starlette 的功能，FastAPI 也支持 WebSocket、Server-Sent Events（SSE）、模板渲染、静态文件等
  4. 误解：Pydantic 模型只能用于请求体 —— 事实：Pydantic 模型也可用于响应模型（Response Model）、配置管理（pydantic-settings）、查询参数模型等

三、锥形深入（西蒙学习法）

集中精力、目标导向、锥形深入——从核心开始，逐步扩展到周边。

第一层：核心基础

安装与启动

# 创建虚拟环境并安装
pip install "fastapi[standard]"

[standard] 包含 uvicorn（ASGI 服务器）、httpx（测试客户端）等常用依赖。创建 main.py 后运行：

# 开发模式（自动重载）
fastapi dev
 
# 生产模式
fastapi run

完整的基础示例

from fastapi import FastAPI
from pydantic import BaseModel
 
app = FastAPI()
 
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool | None = None
 
@app.get("/")
def read_root():
    return {"Hello": "World"}
 
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}
 
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

路径参数与校验

路径参数支持多种数据类型和校验规则：

from fastapi import Path
from typing import Annotated
 
@app.get("/items/{item_id}")
async def read_item(
    item_id: Annotated[int, Path(title="The ID of the item to get", ge=1, le=1000)]
):
    return {"item_id": item_id}

ge=1 表示大于等于 1，le=1000 表示小于等于 1000。无效值会触发自动的 422 错误响应。

查询参数与字符串校验

from fastapi import Query
from typing import Annotated
 
@app.get("/items/")
async def read_items(
    q: Annotated[str | None, Query(min_length=3, max_length=50)] = None,
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(le=100)] = 10,
):
    return {"q": q, "skip": skip, "limit": limit}

请求体 - 嵌套模型

Pydantic 支持多层嵌套的 JSON 结构：

from pydantic import BaseModel
 
class Image(BaseModel):
    url: str
    name: str
 
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list[str] = []
    image: Image | None = None
 
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

请求体 - Field 字段约束

from pydantic import BaseModel, Field
 
class Item(BaseModel):
    name: str = Field(examples=["Foo"])
    description: str | None = Field(
        default=None, title="The description of the item",
        max_length=300
    )
    price: float = Field(gt=0, description="The price must be greater than zero")
    tax: float | None = None

响应模型（Response Model）

用 response_model 参数声明返回类型，实现输出过滤和文档生成：

from fastapi import FastAPI
from pydantic import BaseModel
 
app = FastAPI()
 
class User(BaseModel):
    username: str
    email: str
    full_name: str | None = None
    disabled: bool | None = None
 
class UserOut(BaseModel):
    username: str
    email: str
    full_name: str | None = None
 
@app.get("/users/me", response_model=UserOut)
async def read_user_me():
    # 即使返回包含 disabled 字段，响应中也不会包含它
    return {
        "username": "currentuser",
        "email": "user@example.com",
        "full_name": "Current User",
        "disabled": False,
    }

状态码（Status Codes）

from fastapi import FastAPI, status
from pydantic import BaseModel
 
app = FastAPI()
 
class Item(BaseModel):
    name: str
    description: str | None = None
 
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

错误处理（Error Handling）

from fastapi import FastAPI, HTTPException
 
app = FastAPI()
 
items = {"foo": "The Foo Wrestlers"}
 
@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

HTTPException 不是普通的 Python 异常——它是 FastAPI 专用的，会返回正确的 HTTP 错误响应。

Cookie 和 Header 参数

from fastapi import Cookie, Header, FastAPI
from typing import Annotated
 
app = FastAPI()
 
@app.get("/items/")
async def read_items(
    ads_id: Annotated[str | None, Cookie()] = None,
    user_agent: Annotated[str | None, Header()] = None,
    x_token: Annotated[list[str] | None, Header()] = None,
):
    return {"ads_id": ads_id, "User-Agent": user_agent, "x_token": x_token}

第二层：进阶用法

依赖注入系统（深度）

类作为依赖：不仅函数可以作为依赖，任何可调用对象（包括类）都可以：

from typing import Annotated
from fastapi import Depends, FastAPI
 
app = FastAPI()
 
class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit
 
@app.get("/items/")
async def items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

子依赖（Sub-dependencies）：依赖可以嵌套，形成依赖树：

from typing import Annotated
from fastapi import Depends, FastAPI, Cookie
 
app = FastAPI()
 
def query_extractor(q: str | None = None):
    return q
 
def query_checker(
    extracted_q: Annotated[str, Depends(query_extractor)]
):
    if extracted_q == "admin":
        raise ValueError("Admin queries not allowed")
    return extracted_q
 
@app.get("/items/")
async def read_items(q: Annotated[str, Depends(query_checker)]):
    return {"q": q}

全局依赖：在整个应用或路由器上应用依赖：

from fastapi import Depends, FastAPI, Header, HTTPException
 
async def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")
 
# 所有路由都会先执行 verify_token
app = FastAPI(dependencies=[Depends(verify_token)])

带 yield 的依赖：用于资源管理（如数据库会话）：

from typing import Annotated
from fastapi import Depends, FastAPI
 
app = FastAPI()
 
# 模拟数据库连接
async def get_db():
    db = {"connected": True}
    try:
        yield db
    finally:
        db["connected"] = False  # 清理资源
 
@app.get("/items/")
async def read_items(db: Annotated[dict, Depends(get_db)]):
    return {"db_status": db}

中间件（Middleware）

中间件是在每个请求到达路由之前和响应返回之后执行的函数：

from fastapi import FastAPI, Request
import time
 
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

CORS（跨域资源共享）

前后端分离开发时，必须配置 CORS：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
 
app = FastAPI()
 
# 允许的源列表
origins = [
    "http://localhost:3000",
    "http://localhost:8080",
]
 
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

安全与认证（OAuth2 + JWT）

FastAPI 内置了完整的安全工具链。以下是一个完整的 OAuth2 + JWT 认证示例：

from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from passlib.context import CryptContext
 
# 配置
SECRET_KEY = "your-secret-key-keep-it-secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
app = FastAPI()
 
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
class Token(BaseModel):
    access_token: str
    token_type: str
 
class TokenData(BaseModel):
    username: str | None = None
 
class User(BaseModel):
    username: str
    disabled: bool | None = None
 
class UserInDB(User):
    hashed_password: str
 
# 模拟用户数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "hashed_password": pwd_context.hash("secret"),
        "disabled": False,
    }
}
 
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 authenticate_user(username: str, password: str):
    user_dict = fake_users_db.get(username)
    if not user_dict:
        return False
    user = UserInDB(**user_dict)
    if not verify_password(password, user.hashed_password):
        return False
    return user
 
def create_access_token(data: dict, expires_delta: timedelta | None = None):
    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)
 
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    user_dict = fake_users_db.get(username)
    if user_dict is None:
        raise credentials_exception
    return UserInDB(**user_dict)
 
@app.post("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
        )
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return Token(access_token=access_token, token_type="bearer")
 
@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

数据库集成（SQLAlchemy）

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, Session, DeclarativeBase
 
# 数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
class Base(DeclarativeBase):
    pass
 
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String)
 
Base.metadata.create_all(bind=engine)
 
app = FastAPI()
 
# 数据库会话依赖
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
class UserCreate(BaseModel):
    username: str
    email: str
 
class UserResponse(BaseModel):
    id: int
    username: str
    email: str
 
    class Config:
        from_attributes = True
 
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(username=user.username, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user
 
@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

后台任务（Background Tasks）

from fastapi import FastAPI, BackgroundTasks
 
app = FastAPI()
 
def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message + "\n")
 
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message": "Notification sent in the background"}

WebSocket

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
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 #{client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

文件上传与表单数据

from fastapi import FastAPI, UploadFile, File, Form
from typing import Annotated
 
app = FastAPI()
 
@app.post("/upload/")
async def create_upload_file(
    file: Annotated[UploadFile, File(description="A file to upload")],
    description: Annotated[str, Form()],
):
    contents = await file.read()
    return {
        "filename": file.filename,
        "size": len(contents),
        "description": description,
    }

第三层：深度解析

ASGI 原理与 Starlette

ASGI（Asynchronous Server Gateway Interface）是 WSGI 的异步继任者。关键区别：

特性WSGIASGI
并发模型同步，一次一个请求异步，可同时处理多个请求
WebSocket不支持原生支持
长连接困难原生支持
典型服务器Gunicorn（用于 Flask/Django）Uvicorn（用于 FastAPI/Starlette）

FastAPI 继承自 Starlette，Starlette 是一个轻量级 ASGI 框架。FastAPI 在 Starlette 之上添加了：

  • 基于 Python 类型提示的自动数据校验（通过 Pydantic）
  • 自动 OpenAPI 文档生成
  • 依赖注入系统
  • 安全工具（OAuth2 等）

请求处理流程：

客户端请求
  → Uvicorn（ASGI 服务器）接收
    → Starlette 中间件链
      → FastAPI 依赖注入解析
        → Pydantic 数据校验
          → 路由函数执行
        ← Pydantic 响应序列化
      ← Starlette 中间件链
    ← Uvicorn 返回响应

Pydantic V2 验证引擎

Pydantic V2 是一次从底层重写，核心验证逻辑使用 Rust 编写，性能比 V1 提升约 5-50 倍。

关键特性：

  • BaseModel：声明数据模型，自动进行类型转换和校验
  • 字段约束：Field(gt=0, max_length=100) 等
  • 模型验证器：@field_validator 和 @model_validator
  • 序列化模式：可以定义不同的输入/输出模式
  • model_validate()：从字典创建模型实例（替代 V1 的 parse_obj）
  • model_dump()：将模型转为字典（替代 V1 的 dict()）
from pydantic import BaseModel, field_validator
 
class User(BaseModel):
    name: str
    age: int
    email: str
 
    @field_validator("age")
    @classmethod
    def age_must_be_positive(cls, v):
        if v < 0:
            raise ValueError("Age must be positive")
        return v
 
# 自动类型转换和校验
user = User(name="Alice", age=30, email="alice@example.com")
print(user.model_dump())  # {'name': 'Alice', 'age': 30, 'email': 'alice@example.com'}

OpenAPI 自动生成

FastAPI 根据你的代码自动生成 OpenAPI 3.1 规范（之前是 3.0），这是所有自动文档的基础：

from fastapi import FastAPI
 
app = FastAPI(
    title="My API",
    description="A sample API built with FastAPI",
    version="1.0.0",
    docs_url="/docs",       # Swagger UI 路径
    redoc_url="/redoc",     # ReDoc 路径
    openapi_url="/openapi.json",  # OpenAPI Schema 路径
)

每个路由的以下信息都会被自动记录：

  • HTTP 方法和路径
  • 路径参数、查询参数、请求体
  • 响应模型和状态码
  • 校验规则（长度、范围等）
  • 示例数据

测试策略

FastAPI 使用 TestClient（基于 httpx）进行测试，无需启动真实服务器：

from fastapi.testclient import TestClient
from fastapi import FastAPI
 
app = FastAPI()
 
@app.get("/")
async def read_root():
    return {"msg": "Hello World"}
 
client = TestClient(app)
 
def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

使用 pytest 运行：

pytest test_main.py

覆盖依赖（测试时替换依赖）：

from fastapi.testclient import TestClient
from fastapi import FastAPI, Depends
 
app = FastAPI()
 
async def get_db():
    yield "real_db"
 
async def get_mock_db():
    yield "mock_db"
 
@app.get("/items/")
async def read_items(db: str = Depends(get_db)):
    return {"db": db}
 
def test_with_mock():
    app.dependency_overrides[get_db] = get_mock_db
    client = TestClient(app)
    response = client.get("/items/")
    assert response.json() == {"db": "mock_db"}
    app.dependency_overrides.clear()

部署方案

Docker 部署（推荐）：

Dockerfile：

FROM python:3.14
 
WORKDIR /code
 
COPY ./requirements.txt /code/requirements.txt
 
RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
 
COPY ./app /code/app
 
CMD ["fastapi", "run", "app/main.py", "--port", "80"]

构建和运行：

docker build -t myimage .
docker run -d --name mycontainer -p 80:80 myimage

Uvicorn 多 Worker 模式：

# 生产环境启动多个 worker
fastapi run app/main.py --workers 4 --port 80

关键部署概念：

概念说明常见工具
HTTPSTLS 终止代理处理加密Traefik、Caddy、Nginx
开机启动进程自动启动Docker、Systemd、Kubernetes
自动重启崩溃后自动恢复Docker、Kubernetes、Supervisor
副本复制多进程处理更多请求Uvicorn --workers、K8s
内存管理每个进程独立内存空间根据服务器资源调整 Worker 数量

与其他框架对比

特性FastAPIFlaskDjango REST
异步支持原生 ASGI有限（Flask 2.0+）有限（Django 3.1+）
自动 API 文档内置（Swagger + ReDoc）需第三方（Flask-RESTX）需第三方（drf-spectacular）
类型系统原生 Python 类型提示无无（需手动 Serializer）
数据校验内置（Pydantic）需手动或第三方内置（Serializer）
依赖注入内置强大 DI无无
性能非常高中等中等
ORM 集成框架无关框架无关内置 Django ORM
学习曲线低（熟悉 Python 即可）低中高（需了解 Django）
适合场景API 优先、微服务简单 API、原型全栈应用、CMS

四、要点笔记（康奈尔笔记法）

关键概念速查表

线索/关键词详细笔记
Path Operation用 @app.get/post/put/delete/patch 装饰器声明路由，绑定 HTTP 方法到处理函数
Path ParametersURL 中 {param} 声明，函数参数自动匹配，支持类型转换和校验（Path(ge=1)）
Query Parameters函数参数不在路径中时自动识别，有默认值=可选，无默认值=必需
Request Body用 Pydantic BaseModel 声明 JSON 结构，@app.post/put 使用
Response Modelresponse_model=SomeModel 参数声明返回类型，自动过滤输出字段
Dependency InjectionDepends() 声明依赖，函数/类/生成器均可作依赖，支持嵌套
Middleware@app.middleware("http") 拦截请求/响应，用于日志、CORS、计时等
CORSCORSMiddleware 配置跨域策略：allow_origins/methods/headers
OAuth2 + JWTOAuth2PasswordBearer + PyJWT 实现令牌认证
Background TasksBackgroundTasks.add_task() 执行异步后台操作
WebSocket@app.websocket("/ws") 声明双向通信端点
Pydantic V2Rust 内核校验引擎，BaseModel、Field、field_validator
ASGI异步服务器网关接口，Uvicorn 实现，支持并发、WebSocket
TestClient基于 httpx 的测试客户端，无需启动真实服务器即可测试
FastAPI CLIfastapi dev 开发模式（自动重载），fastapi run 生产模式

核心 API 速查

API / 装饰器用途示例
@app.get(path)声明 GET 路由@app.get("/items/{id}")
@app.post(path)声明 POST 路由@app.post("/items/")
@app.put(path)声明 PUT 路由@app.put("/items/{id}")
@app.delete(path)声明 DELETE 路由@app.delete("/items/{id}")
@app.middleware("http")声明 HTTP 中间件@app.middleware("http")
@app.websocket(path)声明 WebSocket 端点@app.websocket("/ws")
Path()路径参数校验Path(ge=1, le=1000)
Query()查询参数校验Query(min_length=3, max_length=50)
Body()请求体字段约束Body(embed=True)
Field()Pydantic 模型字段约束Field(gt=0, max_length=100)
Depends()声明依赖Depends(get_db)
HTTPException返回 HTTP 错误HTTPException(status_code=404, detail="Not found")
statusHTTP 状态码常量status.HTTP_201_CREATED
UploadFile文件上传处理file: UploadFile
Form()表单数据username: str = Form()
Cookie()Cookie 参数session_id: str = Cookie()
Header()Header 参数user_agent: str = Header()
BackgroundTasks后台任务background_tasks.add_task(func)
TestClient测试客户端client = TestClient(app)
APIRouter()路由分组router = APIRouter(prefix="/api")

本节总结

FastAPI 的设计哲学是类型驱动开发（Type-Driven Development）：

  1. 声明一次，到处生效 —— 一个类型注解同时服务数据校验、文档生成、编辑器支持
  2. 依赖注入是胶水 —— 通过 Depends() 连接数据库、认证、配置等一切组件
  3. Pydantic 是数据核心 —— 所有输入输出都经过 Pydantic 模型的校验和转换
  4. ASGI 是性能基础 —— 原生异步支持让 FastAPI 能处理高并发场景
  5. 标准兼容是保障 —— 基于 OpenAPI 和 JSON Schema，生态工具无缝集成

五、复习与实践（SQ3R · Recite & Review）

核心要点回顾

  1. FastAPI 的核心循环是：类型注解 → 自动校验 → 自动文档 → 自动序列化
  2. Annotated[type, Depends(func)] 是现代 FastAPI 的推荐写法，类型别名（CommonsDep = Annotated[...]）可以减少重复
  3. 依赖注入支持函数、类、生成器（yield），可以嵌套形成依赖树
  4. async def 和 def 可以混用——FastAPI 会正确处理同步/异步调用
  5. 生产部署核心三要素：HTTPS（TLS 代理）、多 Worker（副本复制）、自动重启
  6. 测试时用 dependency_overrides 替换真实依赖，无需 mock 整个框架

动手练习

练习 1：构建一个 CRUD Todo API

要求：

  • GET /todos/ —— 列出所有 todo（支持 skip 和 limit 查询参数）
  • POST /todos/ —— 创建新 todo（使用 Pydantic 模型校验）
  • GET /todos/{todo_id} —— 获取单个 todo（404 处理）
  • PUT /todos/{todo_id} —— 更新 todo
  • DELETE /todos/{todo_id} —— 删除 todo

练习 2：添加认证

在练习 1 的基础上：

  • 实现用户注册和登录（OAuth2 + JWT）
  • 只有认证用户才能访问 todo API
  • 每个用户只能看到自己的 todo

练习 3：Docker 部署

  • 编写 Dockerfile
  • 使用 Docker Compose 同时运行 FastAPI 和 PostgreSQL
  • 配置 CORS 允许前端访问

常见陷阱

  1. 路径参数顺序：/users/me 必须在 /users/{user_id} 之前声明，否则 "me" 会被当作 user_id
  2. 同步阻塞代码：在 async def 中调用同步 IO 操作（如 requests.get()）会阻塞事件循环，应使用 httpx.AsyncClient 或将函数改为 def
  3. Pydantic V1/V2 差异：parse_obj() → model_validate()，.dict() → .model_dump()，class Config → model_config
  4. 忘记 yield 的清理：带 yield 的依赖如果在 yield 之前抛出异常，yield 之后的代码不会执行
  5. CORS 不是后端问题：CORS 是浏览器的安全策略，Postman/curl 不受影响。生产环境中要精确配置 allow_origins，不要用 ["*"]

延伸阅读