本文目录
手写 INSERT INTO items (name, price) VALUES (?, ?) 再 fetchone 转成 dict,字段一多就重复劳动;更麻烦的是改 表 结构时要全局搜字符串 SQL。ORM 的思路相反:先定义 Python Model,让 SQLAlchemy 负责 映射——类对应 表、属性对应列、实例对应行。
映射的核心等式
| 数据库 | ORM 侧 |
|---|---|
表 items | 类 Item |
列 name | 属性 Item.name |
一行 (1, 'book', 9.9) | 实例 Item(id=1, name='book', price=9.9) |
外键 user_id | 关系 Item.owner(可选) |
映射 不是魔法复制:ORM 在 flush/commit 时生成 SQL;读路径上把 cursor 行填进实例。理解这一点,调试 lazy load、N+1 才有抓手。
声明式 Base 与 Model 定义
SQLAlchemy 2.0 声明式风格:
# shop/models/base.py
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass# shop/models/item.py
from datetime import datetime
from sqlalchemy import DateTime, Float, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from shop.models.base import Base
class Item(Base):
__tablename__ = "items"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(120), nullable=False, index=True)
price: Mapped[float] = mapped_column(Float, nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
)要点:
__tablename__指定 表 名;类名Item与表名可不同,但应可预测。Mapped[int]+mapped_column是 2.0 类型 映射 写法,IDE 与 mypy 插件可检查。server_default=func.now()让数据库侧填时间,应用不必每次手设。
首次启动前需建 表(开发可用 Base.metadata.create_all(bind=engine);生产用 Alembic,第 6 篇)。
列类型与约束直觉
常见 映射 选择:
| Python / Mapped | 数据库 | 备注 |
|---|---|---|
Mapped[str] + String(n) | VARCHAR(n) | 长度限制在 DB 层 |
Mapped[int] | INTEGER | 主键、外键 |
Mapped[float] | FLOAT/NUMERIC | 金额更推荐 Numeric |
Mapped[bool] | BOOLEAN | SQLite 用 0/1 |
Mapped[datetime] | TIMESTAMP | 注意 timezone |
Mapped[str | None] | NULL 允许 | 可选字段 |
from decimal import Decimal
from sqlalchemy import Numeric
from sqlalchemy.orm import Mapped, mapped_column
class Product(Base):
__tablename__ = "products"
id: Mapped[int] = mapped_column(primary_key=True)
sku: Mapped[str] = mapped_column(String(32), unique=True)
unit_price: Mapped[Decimal] = mapped_column(Numeric(10, 2))unique=True、index=True、nullable=False 在 Model 上声明,会进入 CREATE TABLE 与 迁移 脚本——保持 Model 为 schema 真相源之一(与 Alembic 协同)。
关系:两张表如何连成对象图
订单与用户是多对一:
# shop/models/user.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from shop.models.base import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True)
items: Mapped[list["Item"]] = relationship(back_populates="owner")# shop/models/item.py 补充
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from shop.models.user import User
class Item(Base):
__tablename__ = "items"
# ... id, name, price 等同上
owner_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
owner: Mapped[User | None] = relationship(back_populates="items")relationship 不自动创建列——外键列 owner_id 仍需显式 映射。back_populates 双向同步,避免只配一边。读 item.owner 时 ORM 可能 lazy 加载另一张 表;在 async 路径要小心(常改 selectinload eager load,第 5 篇涉及 查询 时再细说)。
Model 与 Pydantic Schema 别混
ORM Model 代表持久化行,生命周期绑 Session;Pydantic schema 代表 API 契约。转换示例:
from shop.models.item import Item
from shop.schemas.item import ItemRead
def to_read_model(row: Item) -> ItemRead:
return ItemRead.model_validate(row)ItemRead 需 model_config = {"from_attributes": True}。不要把 Item ORM 类直接当 response_model——懒加载字段、内部列(deleted_at)容易泄露。
注册所有 Model
Alembic 与 create_all 需要 metadata 收集齐:
# shop/models/__init__.py
from shop.models.base import Base
from shop.models.item import Item
from shop.models.user import User
__all__ = ["Base", "Item", "User"]漏 import 的 Model 不会进 metadata,迁移 时对应 表 会悄悄缺失——表现为 autogenerate 认为该表不存在、或运行时查无此表。
映射的加载策略:lazy 与 eager
ORM 默认 lazy:访问 item.owner 时才发 SQL 读 User 表。在同步 Session 里这有时方便;在 FastAPI async 路由 里,lazy 可能在意外时刻触发同步 IO,导致报错或阻塞。加载策略是 映射 行为的一部分,不是事后优化可选项。
常见策略:
| 策略 | 行为 | 适用 |
|---|---|---|
| lazy(默认) | 首次访问关系时 查询 | 简单脚本、确定只在一个 sync 上下文 |
| selectinload | 第二条 IN 查询 批量取关联 | 一对多列表页 |
| joinedload | JOIN 一次取齐 | 一对一、结果集不大 |
设计 Model 关系时就想好列表接口要不要带 owner——这决定 服务 层 查询 怎么写,而不是在 路由 里临时访问关系属性。
继承与多表映射(了解即可)
进阶场景里,一个 Model 类族可以 映射 到多张 表(joined / single table inheritance)。电商里「实物商品 / 虚拟商品」共用部分列、扩展不同列时会出现。起步阶段一个类一张表即可;看到 __mapper_args__ 或抽象基类 Model 时再查文档,不必提前抽象。
时间戳与软删除列
生产 表 常加审计列:
from datetime import datetime
from sqlalchemy import DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
class TimestampMixin:
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
)deleted_at 软删除列不要出现在对外 API schema 里,但 ORM Model 可以有——再次说明 映射 层与契约层分离。软删除的 查询 要在 服务 层统一加 where deleted_at is null,避免某个 路由 忘记过滤。
create_all 与 Alembic 的分工
开发机可以 Base.metadata.create_all(engine) 快速建 表 做实验;团队一旦用 Alembic,应以 迁移 为 schema 真相,Model 改动必须产生 revision。两套并行时,本地 表 可能与同事不一致——新人 onboarding 文档里写清:clone 仓库后先 alembic upgrade head。
复合主键与联合唯一(起步少见)
多数 表 用单列自增 id 即可。关联 表(用户-角色)可能用 (user_id, role_id) 复合主键:
from sqlalchemy import ForeignKey, Integer
from sqlalchemy.orm import Mapped, mapped_column
class UserRole(Base):
__tablename__ = "user_roles"
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), primary_key=True)
role_id: Mapped[int] = mapped_column(ForeignKey("roles.id"), primary_key=True)映射 规则不变:类实例仍对应一行。API 层很少直接把 UserRole 暴露给客户端,它更多是 ORM 表达多对多关系的中间 表。
索引与 table_args
高频 查询 条件应反映在 表 设计上,并在 Model 里可见:
from sqlalchemy import Index
class Item(Base):
__tablename__ = "items"
__table_args__ = (Index("ix_items_name_price", "name", "price"),)
# ... 列定义Alembic autogenerate 对索引的检测不完美,删索引、改索引名时要人工核对 revision。把索引写在 Model 上,是为了让 映射 与 迁移 有共同来源,而不是 DBA 口头通知。
读 ORM 文档时看到 back_populates 与 backref,优先用 back_populates 显式双向——backref 魔法多,大 表 关系网里难追踪。映射 清晰比少写一行字重要。
写 Model 时顺手想 Alembic:加列、改 nullable、加外键,都应能在下一个 revision 里表达。若 表 已在生产,删列往往要分多步 迁移(先停写、再删),Model 注释里记一笔比口头约定可靠。
JSON 列、枚举列在 映射 里常见:Mapped[dict] 配 JSON 类型,或 Python enum.Enum 配 Enum() 列——数据库与 Python 各存一份表示,改枚举值时要同时考虑 迁移 与旧数据。起步用字符串 + CHECK 约束也行,别过早优化。
表 与 Model 命名:复数 表 名 items、单数类名 Item 是常见约定;全项目统一即可。列名 created_at 与属性同名最省心;遗留库 映射 用 mapped_column("legacy_col", key="created_at") 渐进迁移。读 ORM 实例时,你操作的是 Python 属性,SQL 层才是列名——映射 层负责对齐两者。
多对多通过关联 表 Model 表达时,API 往往不直接暴露中间 表 行,而是通过「给用户加角色」这类 服务 用例操作。映射 存在是为了 ORM 能加载关系,不是为了每个 表 都要有 REST 端点。
__repr__ 调试时在 REPL 里一眼认出对象,不影响 映射 到 表:
def __repr__(self) -> str:
return f"<Item id={self.id} name={self.name!r}>"生产日志别打印含 PII 的 repr;开发期 Session 里 print(item) 很有用。
类型注解 Mapped[list["Order"]] 与 relationship 配合,让 mypy 插件能检查关系属性;映射 不仅是运行时,也是静态可读文档。新人读 Model 文件应能还原 ER 图的大致形状,而不必先 \d 数据库 表。
继承 TimestampMixin 等 mixin 类 时,注意 mixin 不要声明 __tablename__——每个 concrete Model 仍对应独立 表。mixin 只复用列 映射,不引入额外 表,这是 ORM 组合优于复制粘贴列定义的方式。
校验 映射 是否与真实 表 一致:本地 alembic upgrade head 后,用数据库客户端对照 Model 列名与 nullable。drift 越早发现,revision 越小;别等生产报错「column does not exist」才补 迁移。团队 review Model PR 时顺带问一句:对应 迁移 在哪。
常见误解
误解一:「ORM 等于不用写 SQL。」 复杂报表、批量更新仍要 SQL 或 Core 表达式;ORM 减的是样板 CRUD。
误解二:「改 Python 属性就改数据库列名。」 属性名与列名可 mapped_column("legacy_name") 解耦。
误解三:「relationship 会自动建外键约束。」 不会;ForeignKey 列必须自己声明。
误解四:「多个 Model 可以 map 同一张表。」 可行(joined table inheritance 等),但属于进阶;起步一对一 表-类。
误解五:「from_attributes 会自动隐藏敏感列。」 它只负责从 ORM 属性拷贝到 schema;哪些字段进 ItemRead 仍由 schema 定义决定,password_hash 别写进去。
误解六:「没有 relationship 就不能 join。」 Core 层 join 仍可用;relationship 只是 ORM 级 映射 便利,不是 SQL 能力上限。
误解七:「Model 字段必须和 API 字段同名。」 持久化名与对外名可不同,靠 schema 做对外映射;Model 服务数据库,schema 服务客户端。
误解八:「主键只能用自增整数。」 UUID、雪花、复合主键都可 映射;选型和分布式、分 表 策略相关,与 ORM 语法无关。
小结
ORM 用声明式 Model 完成 表 到对象的 映射;列、约束、关系在类上表达,Session 负责同步行与实例。改 表 先改 Model 再出 迁移,保持代码与 schema 同源。映射 清晰比花哨技巧更重要。读 Model 应像读简版 ER 图。列类型、nullable、索引、关系四件事写全,后面 Session 与 迁移 才省心。
小结
ORM 用声明式 Model 完成 表 到对象的 映射;列、约束、关系在类上表达,Session 负责同步行与实例。改 表 先改 Model 再出 迁移,保持代码与 schema 同源。映射 清晰比花哨技巧更重要。下一篇讲 Session 生命周期与 CRUD、事务、查询 边界。