本文目录
有了 Engine 和 Model，最容易踩坑的是 Session 边界：对象改了一半没 commit、异常后没 rollback 污染连接池、async 路由里 lazy load 触发隐式 IO。CRUD 本身语法不多，难的是把「一次 HTTP 请求对应哪段 事务、哪条 查询 在何时发」想清楚。
Session 是什么
Session 是 ORM 的工作单元（Unit of Work）：跟踪哪些实例新建、修改、删除，在 flush/commit 时批量生成 SQL。它从 Engine 借 连接，不是长连本身。
同步典型工厂：
# shop/db/session.py
from collections.abc import Generator
from sqlalchemy.orm import Session, sessionmaker
from shop.db.engine import engine
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()异步对应：
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from shop.db.engine import async_engine
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)
async def get_async_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
yield sessionautoflush=False 避免在 查询 前意外 flush；需要时可手动 session.flush()。expire_on_commit=False 减少 commit 后属性访问再发 SQL（async 场景常用）。
挂到 FastAPI：Depends
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from shop.db.session import get_db
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(item_id: int, db: Session = Depends(get_db)) -> dict[str, object]:
item = db.get(Item, item_id)
if item is None:
raise HTTPException(status_code=404, detail="not found")
return {"id": item.id, "name": item.name, "price": item.price}每个请求一个 Session，请求结束 close()，连接 还池。别把 Session 做成全局单例——既不线程安全，也无法按请求 rollback。
CRUD 模式
Create —— 构造实例、加入 session、commit：
from sqlalchemy.orm import Session
from shop.models.item import Item
from shop.schemas.item import ItemCreate
def create_item(db: Session, data: ItemCreate) -> Item:
row = Item(name=data.name, price=data.price)
db.add(row)
db.commit()
db.refresh(row) # 取回 server_default 填充的 id、created_at
return rowRead —— 主键或 查询：
from sqlalchemy import select
def list_items(db: Session, *, offset: int = 0, limit: int = 20) -> list[Item]:
stmt = select(Item).order_by(Item.id).offset(offset).limit(limit)
return list(db.scalars(stmt))Update —— 改已跟踪对象或 bulk update：
def update_item_price(db: Session, item_id: int, new_price: float) -> Item | None:
item = db.get(Item, item_id)
if item is None:
return None
item.price = new_price
db.commit()
db.refresh(item)
return itemDelete：
def delete_item(db: Session, item_id: int) -> bool:
item = db.get(Item, item_id)
if item is None:
return False
db.delete(item)
db.commit()
return Truedb.get(Model, pk) 适合主键；复杂条件用 select + where。2.0 风格统一 session.execute(select(...)) / session.scalars(...)，避免旧式 session.query(Item)（仍可用但不推荐新代码）。
事务：commit、rollback、flush
- flush：把 pending 变更发到 DB，但未 commit，同一 事务 内可见。
- commit：提交 事务，永久化（在隔离级别语义下）。
- rollback：放弃当前 事务，Session 状态重置。
def transfer_credit(db: Session, from_id: int, to_id: int, amount: float) -> None:
try:
sender = db.get(Account, from_id)
receiver = db.get(Account, to_id)
if sender is None or receiver is None:
raise ValueError("account missing")
if sender.balance < amount:
raise ValueError("insufficient funds")
sender.balance -= amount
receiver.balance += amount
db.commit()
except Exception:
db.rollback()
raise业务函数里包 try/rollback 或使用上下文 with db.begin():（2.0 推荐写法之一）。路由层不应吞掉异常而不 rollback，否则下一条请求可能拿到脏 连接。
def create_order(db: Session, item_ids: list[int]) -> Order:
with db.begin():
order = Order()
db.add(order)
for iid in item_ids:
db.add(OrderLine(order=order, item_id=iid))
# begin 块成功则已 commit
db.refresh(order)
return order查询注意事项
- N+1：循环里访问
item.owner会每条发 查询；用selectinload(Item.owner)或joinedload一次 eager load。 - 分页：
limit/offset简单；大偏移考虑 keyset pagination。 - 只读列表：不必 commit；只读也应用
Depends(get_db)保证 Session 关闭。
from sqlalchemy.orm import selectinload
def list_items_with_owner(db: Session) -> list[Item]:
stmt = select(Item).options(selectinload(Item.owner))
return list(db.scalars(stmt))async Session 写法：await session.execute(stmt)、await session.commit()，且不在 async 外 lazy load。
路由里写多少 CRUD
本篇示例把 CRUD 放在 plain 函数里；第 7 篇会把它们挪进 service。路由 只调 item_service.create_item(db, body)，便于单测 mock Session 或内存 SQLite。无论函数放哪，Session 与 事务 契约不变：谁打开、谁提交、谁关闭，应在代码审查里一眼可见。
Session 生命周期与请求边界
一次 HTTP 请求对应一次 Session 是常见默认：请求开始借 连接，请求结束 close，中间无论几次 查询 或一次 commit，都在同一 事务 语境（除非显式 commit 结束当前 事务 并开始新的）。长任务（导出十万行 CSV）不应占着 HTTP Session——应拆后台 job 或流式 查询，否则 连接 池会被慢请求拖干。Web 请求宜短 事务、快 commit。
生成器依赖 yield db 时，FastAPI 在响应发送后执行 finally 关闭 Session；若 路由 里手动 db.commit() 后仍访问已 expire 的对象，可能触发额外 查询 或 DetachedInstanceError——expire_on_commit=False 或 refresh 按场景选用。流式响应（StreamingResponse）生命周期长于普通 JSON 时，更不应长时间占用同一 Session。
批量操作与性能直觉
逐条 create_item 循环 commit 会把 N 次网络往返变成 N 次 事务 fsync，极慢。批量插入可用：
def bulk_create_items(db: Session, rows: list[ItemCreate]) -> None:
db.add_all([Item(name=r.name, price=r.price) for r in rows])
db.commit()更大量数据用 Core insert() 或数据库 COPY。CRUD 教学用单条清晰；上线前对热点路径做 profiling。读路径上 list 接口别在循环里 查询——一次 select + eager load 或分页。
隔离级别与并发（概念）
两个请求同时减库存，若都读到 stock=1 再各减 1，可能超卖。事务 隔离与行锁是数据库层能力；ORM 层常用模式是：
- 乐观：版本号列
version，更新时where id=? and version=?，影响行数 0 则重试。 - 悲观：
select ... for update在 事务 内锁行。
Session 不自动替你选策略；业务逻辑 在 服务 函数里明确。FastAPI 并发高时，这比「多加几个 worker」更关键。
AsyncSession 对照
异步栈里函数签名变为：
async def get_item(session: AsyncSession, item_id: int) -> Item | None:
return await session.get(Item, item_id)
async def list_items(session: AsyncSession) -> list[Item]:
result = await session.scalars(select(Item).limit(20))
return list(result)规则：await session.commit()、禁止在 路由 返回后 lazy load。混用 sync Model 与 async Session 是 2.0 支持的路径，但团队风格要统一，否则 code review 难做。
delete 与级联
删 Model 实例时，ORM 根据 映射 上的 relationship 与 ForeignKey(ondelete=...) 决定关联行怎么办。cascade="all, delete-orphan" 会在删父对象时删子对象；设错了可能一次 commit 删掉半库数据。起步显式删或软删除，确认 业务逻辑 再开 cascade。
def soft_delete_item(db: Session, item_id: int) -> bool:
item = db.get(Item, item_id)
if item is None:
return False
item.deleted_at = func.now() # 需 Model 有 deleted_at 列
db.commit()
return True物理删除与软删除是产品决策，CRUD 函数名应体现语义（remove_item vs archive_item），避免 路由 调用错 服务。
只读查询与写操作分离
列表、详情 查询 不必 commit；但若同一 Session 里先读后写，仍在同一 事务 快照下。只读接口多时可考虑只读 连接 或数据库 replicas——属于部署优化。开发阶段更重要的是：读函数不意外 commit，写函数边界清晰，测试才能稳定。
get 与 scalar_one：db.get(Item, 1) 按主键取；scalars(select(...)).one() 在条件 查询 时期望恰好一行，零行或多行会抛异常——用在「业务上必须唯一」的查询，比手动 if not row 更早暴露错误。
返回 ORM 实例给 路由 前，想清楚 Session 是否仍 open：若在 服务 里 commit 后返回 detached 对象，Pydantic from_attributes 仍可读已加载标量列，但 lazy 关系会炸。常见做法是在 Session 仍活跃时转成 schema，或 eager load 必要关系。
filter 与 where：select(Item).where(Item.price > 10) 是 2.0 写法；动态 查询 可累加 .where(...)。复杂报表仍可能落回 Core 或 raw SQL——Session 不禁止，只是 ORM 糖在简单 CRUD 上更省行。
flush 与 commit 再区分一次：flush 把 pending SQL 发到 DB，但在同一 事务 内仍可 rollback；commit 结束 事务。需要数据库生成的主键在 flush 后可用，而不必 commit——例如 add 后 flush 拿 id，再在同一 事务 写关联表，最后一次性 commit。
嵌套 事务（savepoint）在部分 服务 用例里用 session.begin_nested()；出错只回滚到 savepoint，不拖垮外层。FastAPI 请求级 Session 默认不必 nested，除非单请求内要「试一步、失败则局部撤销」——多数场景一次 commit 更简单。
merge 与 add：merge 把 detached 实例合并进当前 Session，适合从缓存或消息体重建 Model；add 用于新实例。误对已有行 add 可能重复插入；读文档辨清 insert 与 update 路径，CRUD 测试要覆盖更新场景。
计数 查询 用 select(func.count()).select_from(Item)，避免 len(list(scalars(select(Item)))) 把全 表 拉进内存。分页列表只取当前页行数 + total 两次 查询 是常见模式，路由 层只传 page/limit，服务 层组装 查询。
异常后 Session 处于「待 rollback」状态时，继续 查询 可能报 PendingRollbackError——捕获异常后先 db.rollback() 再决定重试或向上抛。FastAPI 全局异常处理器若吞掉异常，别忘记 Session 仍绑在请求上，Depends 的 finally close 前最好显式 rollback 脏 事务。
三种写法怎么选：ORM、Core、Raw
SQLAlchemy 不止「用 Session 操作 Model」一条路，课件里常拆成三层能力——选型错了要么写不动复杂 SQL，要么踩注入：
| 方式 | 典型 API | 适合 | 注意 |
|---|---|---|---|
| ORM | session.add / select(Model) | 单表/少表 CRUD、按对象改字段 | N+1、懒加载 |
| Core 表达式 | insert() / update() / select() 不绑实体 | 批量更新、聚合、复杂 join | 仍参数化，无对象跟踪 |
| Raw SQL | text("SELECT ...") | 方言特有语法、临时排查 | 绝不能拼接用户输入 |
默认优先 ORM；需要「一次改一万行某列」或报表聚合时下沉 Core；只有 ORM/Core 都别扭时才 Raw，且绑定参数：
from sqlalchemy import text
# 危险：f-string 拼进 SQL —— 经典注入面
# bad = text(f"SELECT * FROM items WHERE name LIKE '%{q}%'")
# 正确：命名绑定，驱动负责转义
stmt = text("SELECT id, name FROM items WHERE name LIKE :pattern")
rows = db.execute(stmt, {"pattern": f"%{q}%"}).mappings().all()SQL 注入指攻击者把输入做成「语句的一部分」（如 ' OR 1=1 --），改变查询语义。ORM/select().where(Model.name == q) 与 Core/text(... :param) 走参数绑定，输入只当值；自己用 + / f-string 拼 SQL 才会把值变成代码。搜索框、排序字段名、动态表名都是高危点——表名/列名不能参数化时，必须用白名单枚举，而不是拼接字符串。
选型一句话：简单 CRUD 用 ORM 更顺手；复杂查询混 Core；前两者都难写再 Raw，且始终参数化。耗时外部 I/O 不要夹在同一次持连接的长 事务 里（第 16 篇展开）。
常见误解
误解一：「add 之后数据库里立刻有行。」 通常要 flush/commit；仅 add 可能在内存 pending。
误解二：「commit 失败可以不管。」 必须 rollback 再继续用同一 Session。
误解三：「只读接口不用 Depends Session。」 仍要 Session 做 查询；只是不必 commit。
误解四：「async 路由里用同步 Session 没事。」 会阻塞事件循环；用 AsyncSession 或 run_in_threadpool 包同步 CRUD。
误解五：「一个 Session 可以跨多个请求共享。」 Session 不是线程安全、也不是请求安全；每请求新建、结束关闭是默认正确姿势。
误解六：「rollback 只在上游抛异常时需要。」 主动取消写操作（用户点「放弃」）也应 rollback，避免脏数据留在未提交 事务 里占着锁。
误解七：「list 接口不需要考虑 N+1。」 列表带关联字段时是 N+1 高发区；查询 设计是 CRUD 性能第一课。
误解八：「Depends(get_db) 会自动 rollback。」 finally 里 close 不等于 rollback；异常路径应显式 rollback 再 close，否则池里可能残留未结束 事务 的 连接。
小结
Session 界定 事务 与 ORM 跟踪；CRUD 通过 add/get/select/delete + commit/rollback 完成；查询 注意 eager load 与请求级生命周期。写 CRUD 时先画清「谁 commit、谁 rollback、谁 close」，再谈性能。请求结束务必 close Session，把 连接 还池。单测优先直调 CRUD 函数，少绕 HTTP。出错后先 rollback 再 reuse Session。CRUD 代码是数据层契约，改它要和 迁移、服务 一起 review。养成「写 查询 前先想是否会 N+1」的习惯，比事后加缓存更省事、更稳、更可控。下一篇：迁移 如何管理 表 结构变更，而不是手工改库、手工对账。