K 的一隅

Python FastAPI 后端实战

表变成对象:ORM 映射怎么想

声明式 ORM 如何把数据库表映射为 Python 类:Column、类型、主键与关系的基本直觉,以及和 Pydantic schema 的分工。

9 分钟阅读 更新于 2026-07-28
本文目录

手写 INSERT INTO items (name, price) VALUES (?, ?)fetchone 转成 dict,字段一多就重复劳动;更麻烦的是改 结构时要全局搜字符串 SQL。ORM 的思路相反:先定义 Python Model,让 SQLAlchemy 负责 映射——类对应 、属性对应列、实例对应行。

映射的核心等式

数据库ORM 侧
itemsItem
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 声明式风格:

python
# shop/models/base.py
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass
python
# 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]BOOLEANSQLite 用 0/1
Mapped[datetime]TIMESTAMP注意 timezone
Mapped[str | None]NULL 允许可选字段
python
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=Trueindex=Truenullable=FalseModel 上声明,会进入 CREATE TABLE迁移 脚本——保持 Model 为 schema 真相源之一(与 Alembic 协同)。

关系:两张表如何连成对象图

订单与用户是多对一:

python
# 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")
python
# 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 契约。转换示例:

python
from shop.models.item import Item
from shop.schemas.item import ItemRead


def to_read_model(row: Item) -> ItemRead:
    return ItemRead.model_validate(row)

ItemReadmodel_config = {"from_attributes": True}。不要把 Item ORM 类直接当 response_model——懒加载字段、内部列(deleted_at)容易泄露。

注册所有 Model

Alembic 与 create_all 需要 metadata 收集齐:

python
# 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 查询 批量取关联一对多列表页
joinedloadJOIN 一次取齐一对一、结果集不大

设计 Model 关系时就想好列表接口要不要带 owner——这决定 服务查询 怎么写,而不是在 路由 里临时访问关系属性。

继承与多表映射(了解即可)

进阶场景里,一个 Model 类族可以 映射 到多张 (joined / single table inheritance)。电商里「实物商品 / 虚拟商品」共用部分列、扩展不同列时会出现。起步阶段一个类一张表即可;看到 __mapper_args__ 或抽象基类 Model 时再查文档,不必提前抽象。

时间戳与软删除列

生产 常加审计列:

python
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) 复合主键:

python
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 里可见:

python
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_populatesbackref,优先用 back_populates 显式双向——backref 魔法多,大 关系网里难追踪。映射 清晰比少写一行字重要。

Model 时顺手想 Alembic:加列、改 nullable、加外键,都应能在下一个 revision 里表达。若 已在生产,删列往往要分多步 迁移(先停写、再删),Model 注释里记一笔比口头约定可靠。

JSON 列、枚举列在 映射 里常见:Mapped[dict]JSON 类型,或 Python enum.EnumEnum() 列——数据库与 Python 各存一份表示,改枚举值时要同时考虑 迁移 与旧数据。起步用字符串 + CHECK 约束也行,别过早优化。

Model 命名:复数 items、单数类名 Item 是常见约定;全项目统一即可。列名 created_at 与属性同名最省心;遗留库 映射mapped_column("legacy_col", key="created_at") 渐进迁移。读 ORM 实例时,你操作的是 Python 属性,SQL 层才是列名——映射 层负责对齐两者。

多对多通过关联 Model 表达时,API 往往不直接暴露中间 行,而是通过「给用户加角色」这类 服务 用例操作。映射 存在是为了 ORM 能加载关系,不是为了每个 都要有 REST 端点。

__repr__ 调试时在 REPL 里一眼认出对象,不影响 映射

python
def __repr__(self) -> str:
    return f"<Item id={self.id} name={self.name!r}>"

生产日志别打印含 PII 的 repr;开发期 Sessionprint(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事务查询 边界。