FastAPI 开发

简介

FastAPI 是现代高性能 Python Web 框架，基于 Starlette 和 Pydantic 构建。原生支持 async/await、自动生成 OpenAPI 文档和类型验证，是目前构建 Python API 服务的最佳选择之一。

FastAPI 的设计理念是"让开发者编写尽可能少的代码来实现最多的功能"。它充分利用 Python 的类型注解（type hints）系统——函数参数的类型注解自动成为请求参数的验证规则和 API 文档的描述信息。这种"类型即文档"的设计消除了手动编写 API 文档和参数验证的重复工作。

从技术架构看，FastAPI 由三个核心层构成：(1) Starlette 负责 ASGI 服务器通信、路由匹配和 WebSocket 支持；(2) Pydantic 负责数据模型定义、序列化和运行时验证；(3) FastAPI 本身作为胶水层，将这两者与依赖注入系统结合在一起。理解这个分层有助于在选择中间件和扩展时做出正确决策。

特点

• 1.异步原生 — async/await 高性能
• 2.自动文档 — OpenAPI/Swagger 自动生成
• 3.类型验证 — Pydantic 自动校验请求
• 4.依赖注入 — 内置 DI 系统
• 5.高性能 — 基于 Starlette 的 ASGI 框架
• 6.自动序列化 — JSON 请求/响应自动处理

基础用法

Hello FastAPI

# pip install fastapi uvicorn
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from enum import Enum

app = FastAPI(
    title="User Management API",
    description="用户管理 API 服务",
    version="1.0.0"
)

# 枚举
class UserRole(str, Enum):
    admin = "admin"
    user = "user"
    guest = "guest"

# 请求模型
class UserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=50, description="用户名")
    email: EmailStr
    role: UserRole = UserRole.user
    age: Optional[int] = Field(None, ge=0, le=150)

    class Config:
        json_schema_extra = {
            "example": {"name": "张三", "email": "zhang@example.com", "role": "user", "age": 30}
        }

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    role: str
    age: Optional[int]

# 内存存储
users_db: dict[int, dict] = {}
next_id = 1

@app.get("/")
async def root():
    return {"message": "Welcome to User Management API"}

@app.post("/api/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    global next_id
    user_dict = user.model_dump()
    user_dict['id'] = next_id
    users_db[next_id] = user_dict
    next_id += 1
    return user_dict

@app.get("/api/users", response_model=list[UserResponse])
async def list_users(
    role: Optional[UserRole] = None,
    skip: int = 0,
    limit: int = 10
):
    result = list(users_db.values())
    if role:
        result = [u for u in result if u['role'] == role]
    return result[skip:skip + limit]

@app.get("/api/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    return users_db[user_id]

@app.put("/api/users/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, user: UserCreate):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    user_dict = user.model_dump()
    user_dict['id'] = user_id
    users_db[user_id] = user_dict
    return user_dict

@app.delete("/api/users/{user_id}", status_code=204)
async def delete_user(user_id: int):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    del users_db[user_id]

# 运行: uvicorn main:app --reload --port 8000
# 文档: http://localhost:8000/docs (Swagger UI)
# 文档: http://localhost:8000/redoc (ReDoc)

路由与路径参数详解

from fastapi import FastAPI, Path, Query, Body
from typing import Annotated

app = FastAPI()

# 路径参数（Path）—— 自动类型转换和验证
@app.get("/items/{item_id}")
async def read_item(
    item_id: Annotated[int, Path(title="物品ID", ge=1, description="物品的唯一标识")],
):
    return {"item_id": item_id}

# 查询参数（Query）—— URL 中 ?key=value 的参数
@app.get("/items/")
async def list_items(
    page: Annotated[int, Query(ge=1, default=1, description="页码")],
    size: Annotated[int, Query(le=100, default=10, description="每页数量")],
    search: Annotated[str | None, Query(min_length=1, max_length=50, description="搜索关键词")] = None,
    tags: Annotated[list[str] | None, Query(description="标签过滤")] = None,
):
    return {"page": page, "size": size, "search": search, "tags": tags}

# 请求体（Body）—— POST/PUT 请求的 JSON 体
from pydantic import BaseModel

class ItemUpdate(BaseModel):
    name: str | None = None
    price: float | None = None

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: ItemUpdate,
    importance: Annotated[int, Body(gt=0, embed=True)],  # embed=True 将参数包装在 JSON body 中
):
    return {"item_id": item_id, "item": item, "importance": importance}

# 请求头参数（Header）
from fastapi import Header

@app.get("/headers")
async def read_headers(
    user_agent: Annotated[str | None, Header()] = None,
    accept_language: Annotated[str | None, Header(alias="Accept-Language")] = None,
):
    return {"user_agent": user_agent, "accept_language": accept_language}

# Cookie 参数
from fastapi import Cookie

@app.get("/cookies")
async def read_cookies(
    session_id: Annotated[str | None, Cookie()] = None,
):
    return {"session_id": session_id}

# 表单数据
from fastapi import Form

@app.post("/login")
async def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()],
):
    return {"username": username}

# 文件上传
from fastapi import UploadFile, File

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(content),
    }

@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
    return [{"filename": f.filename, "size": len(await f.read())} for f in files]

依赖注入

DI 系统

from fastapi import Depends, HTTPException, Header
from typing import Annotated

# 数据库会话依赖
def get_db():
    db = {"session": "connected"}
    try:
        yield db
    finally:
        pass  # db.close()

# 认证依赖
async def get_current_user(authorization: str = Header(...)):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="无效的认证头")
    token = authorization[7:]
    # 验证 token
    return {"id": 1, "name": "张三", "role": "admin"}

# 分页依赖
class PaginationParams:
    def __init__(self, page: int = 1, size: int = 10):
        self.page = max(1, page)
        self.size = min(50, max(1, size))
        self.skip = (self.page - 1) * self.size

# 组合依赖
@app.get("/api/me")
async def get_me(user: Annotated[dict, Depends(get_current_user)]):
    return user

@app.get("/api/admin/users")
async def admin_list_users(
    user: Annotated[dict, Depends(get_current_user)],
    pagination: PaginationParams = Depends()
):
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return {"page": pagination.page, "size": pagination.size}

深入依赖注入模式

from fastapi import Depends, HTTPException, status
from typing import Annotated, Generator
from functools import lru_cache

# 1. 配置依赖 —— 使用 yield 的生成器依赖
class DatabaseConfig:
    def __init__(self):
        self.url = "sqlite:///app.db"
        self.pool_size = 10

class Settings:
    def __init__(self):
        self.db = DatabaseConfig()
        self.secret_key = "your-secret-key"
        self.debug = True

@lru_cache()
def get_settings() -> Settings:
    """应用配置（单例）"""
    return Settings()

# 2. 数据库会话依赖
def get_db_session(settings: Annotated[Settings, Depends(get_settings)]) -> Generator:
    """数据库会话依赖"""
    # 模拟数据库连接
    session = {"connection": settings.db.url}
    try:
        yield session
    finally:
        # 确保会话关闭
        session.clear()

# 3. 权限检查依赖 —— 可复用的权限验证
def require_role(role: str):
    """角色权限依赖工厂"""
    async def check_role(
        current_user: Annotated[dict, Depends(get_current_user)]
    ) -> dict:
        if current_user.get("role") != role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"需要 {role} 权限"
            )
        return current_user
    return check_role

# 使用权限工厂
@app.get("/admin/dashboard")
async def admin_dashboard(
    user: Annotated[dict, Depends(require_role("admin"))]
):
    return {"message": f"欢迎管理员 {user['name']}"}

@app.get("/moderator/reviews")
async def moderator_reviews(
    user: Annotated[dict, Depends(require_role("moderator"))]
):
    return {"message": f"欢迎审核员 {user['name']}"}

# 4. 带缓存的依赖
async def get_cached_data(key: str):
    """从缓存获取数据"""
    return {"key": key, "value": "cached_value"}

# 5. 依赖链 —— A 依赖 B，B 依赖 C
async def get_repository(
    db: Annotated[dict, Depends(get_db_session)]
):
    """仓库层依赖"""
    return {"db": db, "type": "repository"}

async def get_service(
    repo: Annotated[dict, Depends(get_repository)]
):
    """服务层依赖"""
    return {"repo": repo, "type": "service"}

@app.get("/data")
async def get_data(
    service: Annotated[dict, Depends(get_service)]
):
    return service

中间件

请求/响应处理

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import time

app = FastAPI()

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

# 计时中间件
@app.middleware("http")
async def timer_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    elapsed = time.time() - start
    response.headers["X-Process-Time"] = f"{elapsed:.3f}s"
    return response

# 全局异常处理
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(status_code=400, content={"detail": str(exc)})

完整的中间件与异常处理

import time
import logging
from fastapi import FastAPI, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import ValidationError

logger = logging.getLogger(__name__)
app = FastAPI()

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=3600,  # 预检请求缓存时间
)

# 请求日志中间件
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    start_time = time.time()
    method = request.method
    path = request.url.path

    logger.info(f"--> {method} {path}")

    try:
        response = await call_next(request)
        elapsed = time.time() - start_time
        logger.info(
            f"<-- {method} {path} "
            f"[{response.status_code}] {elapsed:.3f}s"
        )
        response.headers["X-Process-Time"] = f"{elapsed:.3f}s"
        return response

    except Exception as e:
        elapsed = time.time() - start_time
        logger.error(
            f"<-- {method} {path} "
            f"[ERROR] {elapsed:.3f}s - {str(e)}"
        )
        raise

# 全局异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """处理请求参数验证错误"""
    errors = exc.errors()
    formatted_errors = []
    for error in errors:
        formatted_errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"],
        })
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": "参数验证失败", "errors": formatted_errors},
    )

@app.exception_handler(ValidationError)
async def pydantic_validation_handler(request: Request, exc: ValidationError):
    """处理 Pydantic 模型验证错误"""
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": "数据验证失败", "errors": exc.errors()},
    )

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """全局未捕获异常处理器"""
    logger.error(f"未捕获异常: {exc}", exc_info=True)
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "服务器内部错误"},
    )

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

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message},
    )

后台任务

异步任务

from fastapi import BackgroundTasks

def send_email(to: str, subject: str, body: str):
    print(f"发送邮件至 {to}: {subject}")
    # 实际发送邮件逻辑

@app.post("/api/users", status_code=201)
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks
):
    # 创建用户...

    # 后台发送欢迎邮件
    background_tasks.add_task(
        send_email,
        to=user.email,
        subject="欢迎注册",
        body=f"欢迎 {user.name}！"
    )
    return {"message": "用户创建成功，欢迎邮件已发送"}

高级后台任务模式

import asyncio
from fastapi import FastAPI, BackgroundTasks, HTTPException
from concurrent.futures import ThreadPoolExecutor
import time

app = FastAPI()

# 1. 后台任务类 —— 封装复杂任务逻辑
class EmailService:
    def send_welcome(self, email: str, name: str):
        """发送欢迎邮件（同步，在线程池中运行）"""
        time.sleep(2)  # 模拟 SMTP 请求
        print(f"欢迎邮件已发送至 {email}")

    def send_notification(self, email: str, message: str):
        """发送通知邮件"""
        time.sleep(1)
        print(f"通知已发送至 {email}: {message}")

email_service = EmailService()

# 2. 带依赖注入的后台任务
def process_order_background(order_id: int, db_url: str):
    """后台处理订单"""
    print(f"开始处理订单 {order_id}")
    time.sleep(3)  # 模拟处理
    print(f"订单 {order_id} 处理完成")

@app.post("/orders")
async def create_order(
    background_tasks: BackgroundTasks,
):
    order_id = 12345

    # 添加多个后台任务（按添加顺序执行）
    background_tasks.add_task(
        process_order_background,
        order_id=order_id,
        db_url="sqlite:///app.db",
    )
    background_tasks.add_task(
        email_service.send_notification,
        email="user@example.com",
        message=f"订单 {order_id} 已创建",
    )

    return {"order_id": order_id, "status": "processing"}

# 3. 对于长时间运行的任务，建议使用 Celery 等任务队列
# 而非 FastAPI 的 BackgroundTasks（后者在进程内运行，重启会丢失）

项目结构最佳实践

# 推荐的项目目录结构
#
# my_project/
# +-- app/
# |   +-- __init__.py
# |   +-- main.py              # FastAPI 应用入口
# |   +-- config.py            # 配置管理
# |   +-- dependencies.py      # 全局依赖
# |   +-- models/              # Pydantic 模型
# |   |   +-- user.py
# |   |   +-- order.py
# |   +-- routers/             # 路由模块
# |   |   +-- users.py
# |   |   +-- orders.py
# |   |   +-- auth.py
# |   +-- services/            # 业务逻辑
# |   |   +-- user_service.py
# |   |   +-- order_service.py
# |   +-- repositories/        # 数据访问层
# |   |   +-- user_repo.py
# |   +-- middleware/          # 中间件
# |   |   +-- auth.py
# |   |   +-- logging.py
# |   +-- utils/               # 工具函数
# +-- tests/                   # 测试
# +-- pyproject.toml
# +-- alembic.ini              # 数据库迁移

# app/routers/users.py —— 路由模块示例
from fastapi import APIRouter, Depends, HTTPException
from typing import Annotated

router = APIRouter(prefix="/api/users", tags=["用户"])

@router.get("/", response_model=list[UserResponse])
async def list_users(
    pagination: PaginationParams = Depends(),
):
    """获取用户列表"""
    return []

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(
    user: UserCreate,
    db: Annotated[dict, Depends(get_db_session)],
):
    """创建用户"""
    return user

# app/main.py —— 应用入口
from fastapi import FastAPI
from app.routers import users, orders, auth

app = FastAPI(title="My API", version="1.0.0")

app.include_router(users.router)
app.include_router(orders.router)
app.include_router(auth.router)

测试 FastAPI 应用

import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_create_user():
    """测试创建用户"""
    response = client.post(
        "/api/users",
        json={
            "name": "张三",
            "email": "zhang@example.com",
            "role": "user",
            "age": 30,
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "张三"
    assert "id" in data

def test_get_user_not_found():
    """测试获取不存在的用户"""
    response = client.get("/api/users/99999")
    assert response.status_code == 404

def test_create_user_validation_error():
    """测试参数验证"""
    response = client.post(
        "/api/users",
        json={
            "name": "",  # 名称为空，违反 min_length=1
            "email": "invalid-email",  # 无效邮箱
        },
    )
    assert response.status_code == 422
    errors = response.json()["errors"]
    assert any("name" in e["field"] for e in errors)

def test_list_users_with_pagination():
    """测试分页查询"""
    response = client.get("/api/users?page=1&size=5")
    assert response.status_code == 200
    data = response.json()
    assert isinstance(data, list)

优点

• 1.高性能 — 与 Node.js/Go 相当的性能
• 2.自动文档 — Swagger UI 零配置
• 3.类型安全 — Pydantic 自动验证
• 4.异步支持 — 原生 async/await

缺点

• 1.生态较新 — 社区和插件不如 Flask/Django
• 2.学习曲线 — Pydantic 和 DI 需要理解
• 3.模板引擎 — 不内置模板渲染
• 4.Python 性能 — 仍受限于 Python 运行速度

总结

FastAPI 核心：Pydantic 模型定义请求/响应（自动验证）、路由装饰器（@app.get/post）、依赖注入（Depends）。自动生成 OpenAPI 文档（/docs 和 /redoc）。中间件处理 CORS、计时等。后台任务用 BackgroundTasks。生产部署用 uvicorn + Gunicorn。是构建 Python API 服务的推荐框架，性能和开发体验俱佳。

关键知识点

• Pydantic 模型的 response_model 参数控制响应的序列化和文档
• Depends() 实现依赖注入，yield 形式支持资源清理
• 路径参数、查询参数和请求体通过类型注解自动区分
• 中间件按添加顺序的逆序执行（后添加的先执行）
• TestClient 可以在不启动服务器的情况下测试 API

项目落地视角

• 统一虚拟环境、依赖锁定、格式化和日志方案。
• 把入口、配置、业务逻辑和工具函数拆开，避免单文件膨胀。
• 对网络请求、文件读写和数据处理结果做异常与样本校验。
• 明确项目入口、配置管理、依赖管理、日志和测试策略。
• 使用 APIRouter 按业务域拆分路由，避免 main.py 过大

常见误区

• 把临时脚本直接当生产代码使用。
• 忽略依赖版本、编码、路径和时区差异。
• 只会写 happy path，没有补超时、重试和资源释放。
• 把 notebook 或脚本风格直接带入长期维护项目。
• 在 async 路由中调用同步阻塞函数（应使用 run_in_threadpool）
• 使用 global 变量存储状态（多 worker 部署时数据不一致）
• 忘记配置 CORS 中间件导致前端请求被拒绝

进阶路线

• 学习 Alembic 实现 FastAPI 的数据库迁移
• 掌握 Celery + Redis 实现分布式任务队列
• 研究 OAuth2 + JWT 实现完整的认证授权系统
• 了解 GraphQL 与 FastAPI 的集成

适用场景

• 当你准备把《FastAPI 开发》真正落到项目里时，最适合先在一个独立模块或最小样例里验证关键路径。
• 适合脚本自动化、数据处理、Web 开发和测试工具建设。
• 当需求强调快速迭代和丰富生态时，Python 往往能快速起步。

落地建议

• 统一使用虚拟环境与依赖锁定，避免环境漂移。
• 对核心函数补类型注解、异常处理和日志，减少"脚本黑盒"。
• 一旦脚本进入生产链路，及时补测试和监控。
• 使用 Pydantic v2 的 model_validate/model_dump 替代旧版 API

排错清单

• 先确认当前解释器、虚拟环境和依赖版本是否正确。
• 检查编码、路径、时区和第三方库行为差异。
• 排查同步阻塞、数据库连接未释放或网络请求无超时。
• 确认 Pydantic 模型的验证规则是否合理
• 检查依赖注入链中是否有循环依赖

复盘问题

• 如果把《FastAPI 开发》放进你的当前项目，最先要验证的输入、输出和失败路径分别是什么？
• 《FastAPI 开发》最容易在什么规模、什么边界条件下暴露问题？你会用什么指标或日志去确认？
• 相比默认实现或替代方案，采用《FastAPI 开发》最大的收益和代价分别是什么？

延伸阅读

• 文件操作与异常处理
• Flask Web 开发
• 面向对象编程
• 数据类型与数据结构