本节目标：掌握 Pydantic v2 的字段与自定义校验、FastAPI 依赖注入（含
yield清理）与 SQLAlchemy 2.x 的建模和 CRUD，把三者拼成一个能落库的 API。
适用版本：Python 3.12+（实测 3.14.6）
16.3 数据校验、依赖注入与数据库访问
上一节的接口数据都存在内存字典里，一重启就没了。真实后端要解决三件事：进来的数据必须校验、每个请求都要拿到数据库连接、用完必须还回去。这一节把 Pydantic（9.3 节打过底）、Depends（16.2 节露过脸）和 SQLAlchemy 串起来。安装：pip install "pydantic==2.13.5" "sqlalchemy==2.1.4"。
16.3.1 Pydantic 字段校验
9.3 节讲过 Field 的基本约束，这里补充几个 Web 里最常用的：min_length / max_length（长度）、pattern（正则）、gt / ge（数值下界）。
from pydantic import BaseModel, Field
class Product(BaseModel):
sku: str = Field(min_length=3, pattern=r"^[A-Z]{2}-\d{3}$")
name: str = Field(min_length=1, max_length=40)
price: float = Field(gt=0)
Product(sku="ab-1", name="", price=-1) -> 3 处错误一次报全
sku string_pattern_mismatch String should match pattern '^[A-Z]{2}-\d{3}$'
name string_too_short String should have at least 1 character
price greater_than Input should be greater than 0
Pydantic 会把所有错误一次性收集完再抛，而不是遇到第一个就停——这正是给用户表单做校验时最想要的行为。
16.3.2 自定义校验器：field_validator 与 model_validator
内置约束不够用时自己写。字段级用 field_validator，整模型级用 model_validator（都是 Pydantic v2 写法，不要用 v1 的 @validator）：
from pydantic import BaseModel, field_validator, model_validator
from typing import Self
class DateRange(BaseModel):
start: str
end: str
@field_validator("start", "end")
@classmethod
def not_blank(cls, v: str) -> str:
if not v.strip():
raise ValueError("日期不能为空")
return v.strip()
@model_validator(mode="after")
def check_order(self) -> Self:
if self.start > self.end:
raise ValueError("start 必须早于 end")
return self
DateRange(start="2026-01-01", end="2026-06-30") -> OK
DateRange(start="2026-06-30", end="2026-01-01") -> value_error - start 必须早于 end
model_validator(mode="after") 在字段都校验完之后运行，能读到整个模型，专门做跨字段检查（密码确认、起止日期、金额合计）。返回 Self 能保持类型正确。
16.3.3 嵌套模型、列表模型与 from_attributes
真实数据是嵌套的，模型可以直接嵌：
from pydantic import BaseModel, Field
class OrderLine(BaseModel):
sku: str
qty: int = Field(ge=1)
class Order(BaseModel):
order_id: str
lines: list[OrderLine]
o = Order.model_validate({"order_id": "A1", "lines": [{"sku": "AB-123", "qty": "2"}]})
print(o.lines[0], type(o.lines[0].qty).__name__)
sku='AB-123' qty=2 int
列表里的每个元素都被递归校验成 OrderLine，"2" 也被转成整数。from_attributes 则让模型能从「属性对象」而不是 dict 构造——这是 ORM 对象转 Pydantic 的关键开关，写法是在模型里加 model_config = {"from_attributes": True}，之后就能用 UserOut.model_validate(orm_user) 直接转换。
16.3.4 依赖注入：函数依赖与 yield 清理
Depends 把「准备资源」从端点里抽出来。最实用的形式是带 yield 的依赖：yield 之前是准备，之后是清理，等价于 pytest 的 fixture：
from typing import Annotated
from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
events = []
def get_resource():
events.append("open")
try:
yield "resource" # 交给端点使用
finally:
events.append("close") # 请求结束必定执行
def current_user(res: Annotated[str, Depends(get_resource)]):
events.append("user depends on resource")
return f"user-of-{res}"
app = FastAPI()
@app.get("/me")
def me(user: Annotated[str, Depends(current_user)]):
events.append("handler")
return {"user": user}
print(TestClient(app).get("/me").json())
print("生命周期:", " -> ".join(events))
真实输出：
{'user': 'user-of-resource'}
生命周期: open -> user depends on resource -> handler -> close
顺序很关键：open 最先，close 最后——资源在整个请求处理期间一直可用，响应发出后才释放。数据库连接正是这样管理的：进来时开，处理完关，哪怕端点抛异常也保证关闭（finally 的作用）。这就是「依赖的依赖」：current_user 依赖 get_resource，FastAPI 会按需层层解析。
16.3.5 SQLAlchemy 2.x：引擎、模型与 Session
现在把数据库接进来。SQLAlchemy 2.x 用 DeclarativeBase + Mapped + mapped_column 建模，天然带类型提示：
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
email: Mapped[str] = mapped_column(String(120), unique=True)
engine = create_engine("sqlite:///app.db") # SQLite 零配置，文件落盘
Base.metadata.create_all(engine) # 建表（生产环境交给 Alembic）
三个概念各司其职：Engine 管连接池与方言；模型类描述表结构；Session 是一次工作单元（事务边界）。create_all 只建「不存在的表」，不处理列变更——改字段要迁移工具，见 16.3.11。
16.3.6 select() 2.0 风格 CRUD（真跑 SQLite）
2.0 的核心变化是查询统一走 select() 等构造器，不要再用 1.x 的 session.query()：
from sqlalchemy import select
# Create
with Session(engine) as s:
s.add_all([User(name="Ada", email="ada@x.com"),
User(name="Bob", email="bob@x.com")])
s.commit()
# Read
with Session(engine) as s:
ada = s.execute(select(User).where(User.name == "Ada")).scalar_one()
print("查询:", ada.id, ada.name, ada.email)
# Update
with Session(engine) as s:
bob = s.execute(select(User).where(User.email == "bob@x.com")).scalar_one()
bob.name = "Bobby"
s.commit()
print("更新后:", s.get(User, bob.id).name)
# Delete
with Session(engine) as s:
ada = s.execute(select(User).where(User.name == "Ada")).scalar_one()
s.delete(ada)
s.commit()
真实输出：
查询: 1 Ada ada@x.com
更新后: Bobby
几个要点：s.execute(stmt) 返回 Result，用 .scalar_one()（恰好一个）、.scalars().all()（一列多行）取出对象；s.get(User, id) 按主键取最快；改动字段后要 s.commit() 才落库。
16.3.7 ORM 模型与 Pydantic 模型为何分离
新手常想「让 ORM 模型直接兼当响应模型」，省一份代码。别这么做，有两个硬理由：
- 泄露字段：直接返回 ORM 对象，
__dict__里所有列都会出去。实测把一个含secret列的对象直接返回，得到GET /raw -> 200 {'name': 'Ada', 'secret': 'hashed', 'id': 1}——密码哈希、内部标记全暴露了。而用response_model白名单式输出，只留该给的字段。 - 职责不同：ORM 模型绑定表结构与 Session 生命周期；Pydantic 模型只描述「接口的数据形状」。两者会朝不同方向演化——表加了列不代表接口要暴露它。
正确姿势：数据库实体（ORM）与接口契约（Pydantic）各写各的，在边界处用 from_attributes 转换。
16.3.8 拼起来：FastAPI + Depends + SQLite
把三块拼成完整的 CRUD 接口。注意 Annotated[Session, Depends(get_db)] 这种新写法，比在参数默认值里写 Depends 更清晰：
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
email: Mapped[str] = mapped_column(String(120), unique=True)
engine = create_engine("sqlite:///api.db")
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)
class UserIn(BaseModel):
name: str = Field(min_length=1, max_length=50)
email: str = Field(pattern=r"^[^@]+@[^@]+\.[^@]+$")
class UserOut(BaseModel):
model_config = {"from_attributes": True}
id: int
name: str
email: str
def get_db():
db = SessionLocal()
try:
yield db # 交给端点
finally:
db.close() # 请求结束必定归还连接
DBDep = Annotated[Session, Depends(get_db)]
app = FastAPI()
@app.post("/users", response_model=UserOut, status_code=201)
def create_user(payload: UserIn, db: DBDep):
user = User(name=payload.name, email=payload.email)
db.add(user)
db.commit()
db.refresh(user) # 取回数据库生成的自增 id
return user # ORM 对象 -> UserOut（from_attributes）
@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int, db: DBDep):
user = db.get(User, user_id)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
真实运行结果：
POST /users -> 201 {'id': 2, 'name': 'Bob', 'email': 'bob@x.com'}
GET /users/1 -> {'id': 1, 'name': 'Ada', 'email': 'ada@x.com'}
GET /users/9 -> 404 {'detail': 'User not found'}
POST bad email -> 422
一个请求的完整链条：Pydantic 校验请求体（错则 422）→ 依赖注入开 Session → 端点操作 ORM → response_model 过滤输出 → 依赖的 finally 关闭 Session。每一层各管一段，职责干净。
16.3.9 事务：commit 与 rollback
Session 默认是「延迟提交」：add 只是把对象放进待提交集合，commit() 才真正写库并结束事务；rollback() 丢弃本次所有未提交改动。实测：
回滚后用户数: 1 # add(Carol) 后 rollback()，数据库里仍是 1 条
多条写操作要么全成功、要么全回滚，这是事务的意义。比如转账：扣 A 的钱和加 B 的钱必须在同一个事务里。SQLAlchemy 的 with Session(...) 块退出时若没 commit，会自动回滚，避免脏数据残留。
16.3.10 AsyncSession 与异步数据库
def 端点里用同步 Session 没问题（16.2 节：它跑在线程池）。但 async def 端点里若用同步 Session，会阻塞事件循环。异步数据库用 AsyncSession 配异步驱动：
from sqlalchemy.ext.asyncio import 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_async_db():
async with AsyncSessionLocal() as session:
yield session
@app.get("/users")
async def list_users(db: Annotated[AsyncSession, Depends(get_async_db)]):
result = await db.execute(select(User))
return result.scalars().all()
要点：URL 加驱动后缀（postgresql+asyncpg、sqlite+aiosqlite），查询要 await，expire_on_commit=False 避免提交后访问属性触发额外查询。生产异步数据库一般是 PostgreSQL，SQLite 的异步驱动能力有限。
16.3.11 连接池与迁移工具
连接池参数在 create_engine 里调：pool_size（常驻连接数）、max_overflow（高峰额外连接）、pool_recycle（回收超时连接）、pool_pre_ping=True（借用前探活）。横向扩容时，pool_size × 实例数 不能超过数据库最大连接数。
迁移工具解决 create_all 的短板：它只会建表，不会改表。生产用 Alembic——把「建表/加列/改类型」写成有版本号的迁移脚本，alembic upgrade head 逐步应用、alembic downgrade -1 回退。它的角色类似数据库的 Git：结构变更可追溯、可回滚、可多人协作。
小结
- Pydantic v2 用
Field加约束，field_validator做字段级、model_validator(mode="after")做跨字段校验；错误一次性报全。 - 嵌套与列表模型自动递归校验，
model_config = {"from_attributes": True}让模型能从 ORM 对象构造。 Depends把资源准备抽出来，带yield的依赖在finally里做清理（等价 fixture），并支持依赖的依赖。- SQLAlchemy 2.x 用
DeclarativeBase/Mapped建模，查询走select()，不用 1.x 的session.query()。 - ORM 模型与 Pydantic 模型必须分离：前者管表，后者管接口，直接返回 ORM 对象会泄露字段。
AsyncSession配异步驱动用于async def端点；连接池参数要按实例数规划，结构变更交给 Alembic。
第 16 章到这里完整了：16.1 铺 HTTP 与网关协议的底，16.2 用 FastAPI 写出接口，16.3 让它落库、可校验、可维护。下一章我们离开 Web 服务端，回到自动化脚本的战场——文件批处理、网络爬虫与数据分析，这些正是 Python 最初让人着迷的地方。
阅读导航：上一节：FastAPI 快速上手 · 下一节：文件批处理与办公自动化 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。