本节目标：用 SQLAlchemy 2.x 新风格把数据库表写成带类型标注的 ORM 模型，掌握
select()查询与关系映射，并想清「ORM 模型」和「接口模型」为什么要分层。
适用版本：Python 3.12+（实测 3.14.6）；SQLAlchemy 2.1.4
6.1 SQLAlchemy 2.x ORM 与类型化模型
5.3 节的 FastAPI 服务在关闭时已经能优雅回收资源，但数据还躺在内存字典里。真实后端的第一个数据访问决策是：用不用 ORM、用哪一代风格。这一节不重复 SQL 语法，而是把 SQLAlchemy 2.x 的「类型化建模」讲成一条能直接落进项目的路径。
先明确版本口径：社区口头说的「SQLAlchemy 2.0」指的是 2.x 的新风格（DeclarativeBase、Mapped[]、select()），本机实测版本是 2.1.4。下文所有示例都按 2.1.4 跑过，不再写「2.0.x」。
6.1.1 1.x 风格与 2.x 风格的分界
SQLAlchemy 1.4 起就同时提供两套写法，2.x 把新风格扶正。区别集中在三点：
| 维度 | 1.x 旧风格 | 2.x 新风格 |
|---|---|---|
| 基类 | declarative_base() 工厂函数 | class Base(DeclarativeBase) |
| 字段 | Column(String(50))，类型与属性脱节 | Mapped[str] = mapped_column(String(50)) |
| 查询 | session.query(User).filter(...) | session.scalars(select(User).where(...)) |
新风格的价值在于类型检查器能读懂模型：Mapped[str] 让 mypy 知道 user.name 是 str 而不是 Any，IDE 补全和重构才靠谱。下面是一个最小的两层模型。
6.1.2 定义类型化模型
from __future__ import annotations
from sqlalchemy import ForeignKey, String, create_engine, select
from sqlalchemy.orm import (
DeclarativeBase, Mapped, mapped_column, relationship, Session,
)
class Base(DeclarativeBase):
"""所有模型的公共基类，元数据挂在 Base.metadata 上。"""
class Author(Base):
__tablename__ = "author"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
# 一对多：作者有多本书
books: Mapped[list["Book"]] = relationship(back_populates="author")
class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(100))
price: Mapped[int]
author_id: Mapped[int] = mapped_column(ForeignKey("author.id"))
author: Mapped[Author] = relationship(back_populates="books")
engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)
几个容易踩的点：
Mapped[int]不带默认值时，主键列会被推断为NOT NULL；非空约束跟着类型走。- 关系两端都要写，
back_populates把Author.books和Book.author绑定成一对镜像，双向赋值会自动同步。 Mapped[list["Book"]]里的字符串前向引用配合文件顶部的from __future__ import annotations，解决两个类互相引用的顺序问题。- 只有
price: Mapped[int]没写mapped_column也能建列，SQLAlchemy 会按类型推导；要指定长度、索引、默认值时才显式调用。
6.1.3 插入与查询
建表之后先灌数据，再按 select() 查询。注意 session.scalars() 返回的是 ORM 对象本身，不是 Row 元组。
with Session(engine) as session:
a1, a2 = Author(name="Leeting"), Author(name="Ada")
session.add_all([a1, a2])
session.flush() # 拿到自增主键，但还没提交
session.add_all([
Book(title="Py 实战", price=59, author_id=a1.id),
Book(title="Py 入门", price=49, author_id=a1.id),
Book(title="计算纲要", price=99, author_id=a2.id),
])
session.commit()
with Session(engine) as session:
stmt = select(Book).where(Book.price >= 50).order_by(Book.price.desc())
for book in session.scalars(stmt):
print(book.id, book.title, book.price)
实测输出（SQLite，SQLAlchemy 2.1.4）：
3 计算纲要 99
1 Py 实战 59
select(Book) 只取 Book 一张表；.where() 接受的是列表达式而不是字符串，所以 Book.price >= 50 由 Python 直接写、类型检查器能校验，没有拼 SQL 字符串的注入风险。
6.1.4 更多查询形态
真实项目里远不止按条件取行，还有主键直取、联表、聚合和分页。下面这几段都在同一份数据上跑过。
with Session(engine) as session:
# 按主键取单个对象，命中身份映射时不再发 SQL
print("get:", session.get(Book, 1).title)
# 联表：按关联表的字段过滤
stmt = (select(Book).join(Author)
.where(Author.name == "Leeting").order_by(Book.price))
print("join:", [(b.title, b.price) for b in session.scalars(stmt)])
# 聚合：分组统计作者的书数与总价
stmt = (select(Author.name, func.count(Book.id), func.sum(Book.price))
.join(Book).group_by(Author.name))
print("agg:", session.execute(stmt).all())
# 计数与分页
print("count:", session.scalar(select(func.count()).select_from(Book)))
page = session.scalars(select(Book).order_by(Book.id).limit(2).offset(0)).all()
print("page1:", [b.title for b in page])
实测输出：
get: Py 实战
join: [('Py 入门', 49), ('Py 实战', 59)]
agg: [('Ada', 1, 99), ('Leeting', 2, 108)]
count: 3
page1: ['Py 实战', 'Py 入门']
几个要点：
session.get(Model, pk)优先于select().where(pk==...)：它先查身份映射，同一个 session 内命中缓存就不再发 SQL。join()的对象是 ORM 实体，SQLAlchemy 会按外键自动推断ON条件，不用手写book.author_id = author.id。- 聚合查询用
session.execute()而非scalars()：前者返回Row元组，正好匹配多列结果；scalars()只取第一列。 - 分页必须带
order_by：没有稳定排序时limit/offset的结果顺序不确定，翻页会漏行或重复。 func.count()、func.sum()等聚合函数来自sqlalchemy.func，会被翻译成对应的 SQL 方言函数。
6.1.5 加载策略：惰性 vs 预加载
关系字段默认是惰性加载（lazy）：访问 author.books 的那一刻才发第二条 SQL。这在单条记录上没问题，一旦在循环里访问就退化成 N+1（6.2 节专门治理）。想提前加载有两种常见选择：
from sqlalchemy.orm import selectinload, joinedload
# 方案 A：selectinload —— 额外发一条 WHERE author_id IN (...) 的查询
authors = session.scalars(
select(Author).options(selectinload(Author.books))
).all()
# 方案 B：joinedload —— 用 LEFT OUTER JOIN 一条 SQL 取回，结果需去重
authors = session.scalars(
select(Author).options(joinedload(Author.books))
).unique().all()
两者的取舍：
| 策略 | SQL 条数 | 结果形状 | 适用 |
|---|---|---|---|
selectinload | 2（主表 + IN 查询） | 无重复行 | 一对多、多对多，推荐默认 |
joinedload | 1（JOIN） | 有笛卡尔膨胀，需 .unique() | 多对一、一对一 |
| 惰性 | 1 + N | 按需 | 只读单条时够用 |
在 SQLite 上对 10 个作者、50 本书的实测：惰性 11 条 SQL、selectinload 2 条、joinedload 1 条。数字本身不重要，重要的是加载策略是模型层就能声明的工程决策，不必等到线上慢查询才回头补。
6.1.6 ORM 模型与接口模型必须分层
新手常把 ORM 对象直接当响应体返回，这在工程上是隐患：ORM 对象带着 author_id、关系代理、甚至未加载的惰性字段，一旦 Pydantic 尝试序列化就会触发意外查询，也可能把不该暴露的列吐给客户端。
正确做法是两套模型、显式转换：
from pydantic import BaseModel, ConfigDict
class BookOut(BaseModel):
model_config = ConfigDict(from_attributes=True) # 允许从 ORM 对象构造
id: int
title: str
price: int
def to_book_out(book: Book) -> BookOut:
return BookOut.model_validate(book)
from_attributes=True 让 Pydantic 按属性名从任意对象读取字段。分层之后，数据库 schema 的变更和 API 契约的变更就能各自演进——加一列不会自动漏到接口，删一列不会让接口崩。7.1 节会把接口模型的设计展开。
6.1.7 建模时的几个工程约定
- 表名单数、小写下划线：
author/book，不要AuthorTable。 - 主键统一用代理键：自增整数或 UUID，别拿业务字段当主键。
- 时间列放数据库侧默认：
server_default=func.now()比 Python 侧datetime.now()更稳，避免多实例时钟漂移。 - 约束尽量落到库上：
nullable=False、unique=True、外键，能交给数据库的校验不要只写在应用层。 - 不要
create_all上生产：它只建不改，结构演进交给 6.3 节的 Alembic。
6.1.8 用内存 SQLite 测试模型层
4.1 节讲过 fixture 分层，落到数据访问层就是一个「每个用例一套干净 schema」的 fixture。内存 SQLite 建库几乎零成本，非常适合单测：
import pytest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
@pytest.fixture
def session() -> Session:
engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)
with Session(engine) as s:
yield s
def test_insert_and_query(session: Session) -> None:
session.add(Book(title="Py 实战"))
session.commit()
assert session.scalar(select(Book.title)) == "Py 实战"
实测（pytest 9.1.1）：
1 passed in 0.72s
要点：
:memory:是进程内、随连接销毁的库，天然隔离，无需清表。- fixture 里
yield出 session，用例结束自动close（对应 5.2 节的依赖清理思路）。 - 想验证真实数据库方言（如 PG 的
JSONB），把 fixture 换成 4.2 节的容器化测试环境即可，模型代码一行不用改。
延伸阅读：Python 数据库与 ORM 完全指南 从选型讲到异步驱动，本节聚焦的是「新风格建模怎么落进项目」。
小结
- SQLAlchemy 2.1.4 的新风格用
class Base(DeclarativeBase)取代declarative_base()，用Mapped[]让模型带上可被类型检查器读懂的标注。 - 字段写成
Mapped[str] = mapped_column(String(50))，主键、非空、长度都跟着类型与参数走。 - 查询统一走
select()+session.scalars()，条件用 Python 表达式而非 SQL 字符串。 - 关系用
relationship+back_populates双向绑定，加载策略（惰性 /selectinload/joinedload）在模型层就能声明。 - ORM 模型管表、Pydantic 模型管接口，必须分层并用
from_attributes显式转换，不要直接返回 ORM 对象。 create_all只用于测试和原型，生产结构演进交给迁移工具。
模型能建、能查了，但「一次请求里 session 开在哪、事务边界划在哪、连接池够不够、查询为什么慢」这些运行时问题还没碰。下一节把 session 生命周期、事务、连接池和 N+1 治理一次讲透。
阅读导航：上一节：生命周期、优雅关闭与健康探针 · 下一节：事务、连接池与 N+1 治理 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。