FastAPI 学习总结 - 第二部分：进阶特性

六、数据验证

Pydantic 模型

Pydantic 是 FastAPI 的核心数据验证库，使用 Python 类型提示进行数据验证和序列化。

基本用法：

from pydantic import BaseModel, Field
from typing import Optional, List

class User(BaseModel):
    """用户模型"""
    username: str = Field(..., description="用户名", min_length=3, max_length=50)
    email: str = Field(..., description="邮箱")
    age: int = Field(18, description="年龄", ge=0, le=120)
    is_active: bool = Field(True, description="是否活跃")
    tags: List[str] = Field([], description="标签列表")
    bio: Optional[str] = Field(None, description="个人简介", max_length=500)

Field 参数说明：

参数 说明 示例
default 默认值 Field(18)
... 必填字段 Field(...)
ge 大于等于 Field(..., ge=0)
gt 大于 Field(..., gt=0)
le 小于等于 Field(..., le=100)
lt 小于 Field(..., lt=100)
min_length 最小长度 Field(..., min_length=3)
max_length 最大长度 Field(..., max_length=50)
pattern 正则匹配 Field(..., pattern="^[a-z]+$")
description 字段描述 Field(..., description="用户名")

自定义验证器

使用 @validator 装饰器添加自定义验证逻辑：

from pydantic import BaseModel, validator

class User(BaseModel):
    username: str
    email: str
    
    @validator('username')
    def username_must_be_alphanumeric(cls, v):
        """验证用户名只能包含字母和数字"""
        if not v.isalnum():
            raise ValueError('用户名只能包含字母和数字')
        return v
    
    @validator('email')
    def email_must_contain_at(cls, v):
        """验证邮箱格式"""
        if '@' not in v:
            raise ValueError('邮箱格式不正确')
        return v

查询参数校验

使用 Query 校验查询参数：

from fastapi import Query

@app.get("/search")
def search(
    q: str = Query(
        None, 
        description="搜索关键词",
        min_length=3, 
        max_length=50,
        regex="^[a-zA-Z0-9]+$"
    ),
    page: int = Query(1, description="页码", ge=1),
    page_size: int = Query(10, description="每页数量", ge=1, le=100)
):
    """
    搜索接口
    
    :param q: 搜索关键词
    :param page: 页码（默认1）
    :param page_size: 每页数量（默认10）
    :return: 搜索结果
    """
    return {
        "q": q,
        "page": page,
        "page_size": page_size,
        "results": []
    }

路径参数校验

使用 Path 校验路径参数：

from fastapi import Path

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(
        ..., 
        description="商品ID",
        gt=0, 
        le=1000
    )
):
    """
    获取商品详情
    
    :param item_id: 商品ID（1-1000）
    :return: 商品信息
    """
    return {"item_id": item_id}

嵌套模型

Pydantic 支持嵌套模型：

from pydantic import BaseModel
from typing import List

class Address(BaseModel):
    """地址模型"""
    street: str
    city: str
    zipcode: str

class User(BaseModel):
    """用户模型"""
    username: str
    email: str
    addresses: List[Address] = []  # 嵌套列表

@app.post("/users")
def create_user(user: User):
    return user.dict()

请求示例：

{
    "username": "john",
    "email": "john@example.com",
    "addresses": [
        {"street": "123 Main St", "city": "Beijing", "zipcode": "100000"},
        {"street": "456 Oak Ave", "city": "Shanghai", "zipcode": "200000"}
    ]
}

七、异常处理

HTTPException

使用 HTTPException 抛出自定义异常：

from fastapi import HTTPException

# 模拟数据库
items_db = {
    1: {"name": "Item 1", "price": 10},
    2: {"name": "Item 2", "price": 20}
}

@app.get("/items/{item_id}")
def read_item(item_id: int):
    """
    获取商品详情
    
    :param item_id: 商品ID
    :return: 商品信息
    """
    item = items_db.get(item_id)
    if not item:
        raise HTTPException(
            status_code=404,
            detail=f"商品 {item_id} 不存在",
            headers={"X-Error": "ItemNotFound"}
        )
    return item

响应示例（404）：

{
    "detail": "商品 3 不存在"
}

自定义异常处理器

使用 @app.exception_handler 自定义异常处理：

from fastapi import Request
from fastapi.responses import JSONResponse

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

# 自定义异常处理器
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.message}
    )

@app.get("/custom-error")
def raise_custom_error():
    raise CustomException(
        status_code=400,
        message="自定义错误信息"
    )

全局异常处理器

处理所有未捕获的异常：

from fastapi import Request
from fastapi.responses import JSONResponse
import traceback

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """处理所有未捕获的异常"""
    return JSONResponse(
        status_code=500,
        content={
            "error": "服务器内部错误",
            "detail": str(exc),
            "traceback": traceback.format_exc()
        }
    )

八、响应处理

响应模型

使用 response_model 指定响应格式：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserIn(BaseModel):
    """请求模型"""
    username: str
    password: str
    email: str

class UserOut(BaseModel):
    """响应模型"""
    username: str
    email: str

@app.post("/users", response_model=UserOut)
def create_user(user: UserIn):
    """
    创建用户（不返回密码）
    
    :param user: 用户信息
    :return: 用户信息（隐藏密码）
    """
    # 实际应用中会保存到数据库
    return user

请求示例：

{
    "username": "john",
    "password": "secret",
    "email": "john@example.com"
}

响应示例（密码被过滤）：

{
    "username": "john",
    "email": "john@example.com"
}

响应模型配置

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = 10.5
    tags: list = []

items_db = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2}
}

@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: str):
    return items_db[item_id]

响应示例（包含默认值）：

{
    "name": "Foo",
    "description": null,
    "price": 50.2,
    "tax": 10.5,
    "tags": []
}

排除未设置的字段

使用 response_model_exclude_unset 只返回实际设置的字段：

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
def read_item(item_id: str):
    return items_db[item_id]

响应示例（不包含默认值）：

{
    "name": "Foo",
    "price": 50.2
}

包含/排除特定字段

@app.get("/items/{item_id}", response_model=Item, response_model_include={"name", "price"})
def read_item(item_id: str):
    return items_db[item_id]

@app.get("/items/{item_id}", response_model=Item, response_model_exclude={"tax"})
def read_item(item_id: str):
    return items_db[item_id]

自定义响应状态码

from fastapi import status

@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item(name: str):
    """
    创建商品（返回201状态码）
    
    :param name: 商品名称
    :return: 商品信息
    """
    return {"name": name}

自定义响应头

from fastapi import Response

@app.get("/custom-header")
def custom_header(response: Response):
    """
    设置自定义响应头
    
    :param response: 响应对象
    :return: 消息
    """
    response.headers["X-Custom-Header"] = "custom-value"
    return {"message": "Hello"}

九、依赖注入

基本依赖

依赖注入允许我们将重复的逻辑抽取为独立的函数：

from fastapi import Depends, FastAPI

app = FastAPI()

def get_db():
    """获取数据库连接"""
    db = "数据库连接对象"
    print("建立数据库连接")
    return db

@app.get("/items")
def read_items(db=Depends(get_db)):
    """
    获取商品列表
    
    :param db: 数据库连接（通过依赖注入）
    :return: 商品列表
    """
    return {"db": db, "items": []}

@app.get("/users")
def read_users(db=Depends(get_db)):
    """
    获取用户列表
    
    :param db: 数据库连接（通过依赖注入）
    :return: 用户列表
    """
    return {"db": db, "users": []}

带 yield 的依赖（资源管理）

使用 yield 实现资源的获取和释放：

from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    """
    获取数据库会话
    
    使用 yield 确保会话正确关闭
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/items")
def read_items(db: Session = Depends(get_db)):
    """
    获取商品列表
    
    :param db: 数据库会话
    :return: 商品列表
    """
    items = db.query(Item).all()
    return items

多级依赖

依赖可以相互依赖：

from fastapi import Depends

def get_config():
    """获取配置"""
    return {"secret_key": "secret123"}

def get_db(config=Depends(get_config)):
    """获取数据库连接（依赖配置）"""
    db = f"数据库连接使用密钥: {config['secret_key']}"
    return db

@app.get("/items")
def read_items(db=Depends(get_db)):
    """
    获取商品列表
    
    :param db: 数据库连接
    :return: 商品列表
    """
    return {"db": db}

带参数的依赖

依赖函数可以接受参数：

from fastapi import Depends, Query

def pagination(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    """分页参数依赖"""
    return {"skip": skip, "limit": limit}

@app.get("/items")
def read_items(pagination_params=Depends(pagination)):
    """
    获取商品列表（分页）
    
    :param pagination_params: 分页参数
    :return: 商品列表
    """
    items = [{"id": i, "name": f"Item {i}"} for i in range(100)]
    skip = pagination_params["skip"]
    limit = pagination_params["limit"]
    return items[skip : skip + limit]

类作为依赖

使用类作为依赖：

from fastapi import Depends

class Database:
    def __init__(self):
        self.connection = "数据库连接"
    
    def query(self, sql):
        """执行查询"""
        return f"执行SQL: {sql}"

def get_db():
    """获取数据库实例"""
    db = Database()
    try:
        yield db
    finally:
        # 清理资源
        pass

@app.get("/query")
def query_db(db: Database = Depends(get_db)):
    """
    查询数据库
    
    :param db: 数据库实例
    :return: 查询结果
    """
    result = db.query("SELECT * FROM users")
    return {"result": result}

全局依赖

为整个应用添加全局依赖：

from fastapi import Depends, FastAPI

app = FastAPI(dependencies=[Depends(get_db)])

@app.get("/items")
def read_items():
    """获取商品列表（自动注入数据库依赖）"""
    return {"items": []}
# fastapi # python # 服务器 # 开发语言
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

当 AI 编程从“辅助”走向“治理”：一个前端工程师的 2026 年观察

2026 年，AI 编程正在从代码补全、AI IDE 进入 Agent 工程化阶段。它不再只是帮开发者写几段代码，而是开始参与 issue 处理、测试修复、重构迁移甚至 PR 提交。但真正的问题也随之变化：AI 是否能生成代码，已经不是核心；更关键的是，它生成的代码是否安全、可维护、可审计。本文结合 METR 开发者生产力研究、Agentic Coding 趋势、MCP 安全风险，以及实际开发中遇

avatar AtomGit开源社区

编程开挂指南：构建你的超能力技能树

《代码开挂指南：IT人的超能力培养体系》摘要 该体系构建了从基础到前沿的立体化技能树：核心层涵盖编程语言、算法、设计模式三大支柱；工具链维度要求精通开发工具、版本控制和调试技术；工程化思维强调规范流程与自动化部署。认知层面培养代码直觉与抽象建模能力，通过快捷键、自动化等技巧提升效能。前瞻性拓展包括AI编程、云原生等未来技术，建议通过技术雷达跟踪、开源贡献和写作输出来持续进化。每个技能点均配备效率对

avatar AtomGit开源社区
cover

Cursor 对接国内大模型（DeepSeek 等）完整部署配置教程

avatar AtomGit开源社区