第十二章:实战场景
12.1 场景一:FastAPI 接口开发
Pydantic 是 FastAPI 的基石,请求体、响应体、查询参数全靠它:
python
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class CreateUserReq(BaseModel):
name: str = Field(min_length=1, max_length=20)
age: int = Field(ge=0, le=150)
class UserResp(BaseModel):
id: int
name: str
@app.post("/users", response_model=UserResp)
def create_user(req: CreateUserReq):
# 走到这里,req 已经是校验过的干净数据,直接用
return UserResp(id=1, name=req.name)要点:response_model 会自动过滤多余字段(比如忘删的内部字段),是最后一道安全网。
12.2 场景二:应用配置管理
用 pydantic-settings 管理配置,环境变量自动加载、类型自动校验:
bash
pip install pydantic-settingspython
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
redis_host: str = "localhost"
debug: bool = False
class Config:
env_file = ".env" # 自动读取 .env 文件
settings = Settings() # 自动从环境变量 / .env 读取并校验
print(settings.database_url)配置错了(比如缺了 DATABASE_URL)程序启动就报错,而不是运行到一半才炸——fail fast,早死早超生。
12.3 场景三:清洗脏数据
处理外部数据(爬虫、第三方 API、Excel 导入)时的典型组合拳:
python
from pydantic import BaseModel, field_validator, ValidationError
class ProductRow(BaseModel):
name: str
price: float
@field_validator("price", mode="before")
@classmethod
def parse_price(cls, v):
# 处理 "¥19.9"、"19.9元" 这类脏数据
if isinstance(v, str):
cleaned = v.replace("¥", "").replace("元", "").strip()
return float(cleaned)
return v
valid_rows, bad_rows = [], []
for raw in messy_data:
try:
valid_rows.append(ProductRow.model_validate(raw))
except ValidationError as e:
bad_rows.append((raw, e.errors())) # 记下哪行错、为什么错12.4 场景四:LLM 结构化输出
python
class ArticleSummary(BaseModel):
title: str
key_points: list[str] = Field(max_length=5)
sentiment: Literal["positive", "neutral", "negative"]
# 把 ArticleSummary.model_json_schema() 交给 LLM,
# 拿到输出后 ArticleSummary.model_validate_json(llm_output) 兜底校验