12 ORM 介绍 —— 你的 Python 对象和 MySQL 表之间，隔着一层什么东西？

前置阅读: 04-路由
关键词: ORM, aiomysql, 异步映射, 数据访问层
难度: ★★☆☆☆

场景导入：为什么不能直接把 dict 存进 MySQL？

前 11 章中，图书端点的数据一直存在内存里——进程重启就丢了。真实的图书管理 API 需要把书目持久化到 MySQL，但问题来了：Python 里操作的是 Book 对象（有属性、有类型、有关系），MySQL 里存的是行和列（有主键、有外键、有索引）。这两套模型之间的"翻译工作"如果全用手写 SQL 完成，每个查询都要手动管理连接、拼参数、读游标、转字典——样板代码很快就会淹没业务逻辑。

这就是 ORM（Object-Relational Mapping，对象关系映射）要解决的问题。它在 Python 对象和数据库行之间建立对应关系，让你用 select(Book).where(Book.isbn == isbn) 而不是 SELECT id, title, isbn FROM books WHERE isbn = %s 来操作数据。但 ORM 不是"不用懂 SQL"的借口——它只是一层翻译器，理解它生成的 SQL 仍然是你必须掌握的能力。

本篇从概念层面把 aiomysql 原生 SQL 和 SQLAlchemy 异步 ORM 两条路径并排对比，为第 13 章的异步引擎与会话栈铺垫。

原理解析：两条路径，同一个目的地

对象关系映射的核心思想其实很朴素：MySQL 的 books 表映射为 Python 的 Book 类，一行数据对应一个 Book 实例，列值对应实例属性。ORM 的责任就是在这两种表示之间来回翻译——查询时把 SQL 结果集变成对象列表，写入时把对象属性变成 INSERT 语句。

但翻译这件事可以发生在不同的抽象层次上。下面这张图把原生 SQL 路径和 ORM 路径并排画了出来：

flowchart LR subgraph Native["aiomysql 原生 SQL 路径"] A["应用代码"] --> B["await aiomysql.connect"] B --> C["参数化 SQL 与参数元组"] C --> D["await cur.execute"] D --> E["(MySQL)"] E --> F["await cur.fetchone"] F --> G["手动转换字典"] end subgraph ORM["SQLAlchemy 异步 ORM 路径"] H["应用代码"] --> I["select 查询表达式"] I --> J["await db.execute"] J --> K["映射层生成参数化 SQL"] K --> L["aiomysql 驱动"] L --> M["(MySQL)"] M --> N["映射为 Book 对象"] end

（图注：两条路径最终都通过 aiomysql 异步驱动访问 MySQL，区别在于连接生命周期和结果映射由谁管理——你手动写，还是 ORM 会话统一协调。）

两者的关键差异不在于"有没有 SQL"，而在于控制权和便利性的交换：

sequenceDiagram participant App as 应用代码 participant ORM as SQLAlchemy ORM participant Driver as aiomysql participant DB as MySQL alt 原生 SQL App->>Driver: await aiomysql.connect() Driver->>DB: 异步建立连接 DB-->>Driver: 连接可用 Driver-->>App: 返回异步连接 App->>Driver: await cur.execute(SQL, params) Driver->>DB: 发送参数化 SQL DB-->>Driver: 返回结果行 App->>Driver: await cur.fetchone() Driver-->>App: 返回单行数据 else ORM App->>ORM: await db.execute(select(...)) ORM->>ORM: 编译表达式并绑定参数 ORM->>Driver: 执行参数化 SQL Driver->>DB: 发送语句 DB-->>Driver: 返回结果行 Driver-->>ORM: 返回游标结果 ORM-->>App: 返回映射结果 end

（图注：原生路径需要你显式管理"连接 → 游标 → 执行 → 读取 → 关闭"全流程；ORM 路径你只需要声明查询意图，连接调度和结果包装由会话协调。）

ORM 的设计动机可以从三个角度理解：

减少重复样板。 每个查询都手工执行"取得连接、创建游标、绑定参数、读取行、构造对象"这一套流程，业务规则会被基础设施代码淹没。ORM 把这些动作收敛到会话层，查询代码更接近领域意图——select(Book).where(Book.isbn == isbn) 读起来像是在描述"我要那本 ISBN 匹配的书"，而不是在指挥数据库。

统一参数绑定。 原生 SQL 和 ORM 都必须把用户输入和 SQL 模板分离——这是防注入的底线。ORM 不会让注入风险自动消失，但表达式 API 减少了误用字符串格式化的机会。Book.isbn == isbn 被 SQLAlchemy 重载为 BinaryExpression，最终翻译为 WHERE isbn = %s 加参数绑定——你不会不小心写成 f"WHERE isbn = '{isbn}'"。

集中管理对象状态。 ORM 通过身份映射保证同一会话内，相同主键通常对应同一个 Python 对象——你从两个不同查询拿到同一个 book_id=1 的 Book，它们是同一个实例。工作单元则追踪所有变更，在 commit 时一并 flush。但这种便利也带来学习成本：你仍然需要理解事务、索引、连接池和实际生成的 SQL——ORM 是数据库知识的放大器，不是替代品。

代码实现：同一个查询，两种写法

下面用同一个 ISBN 查询对比两条异步路径。原生版本展示完整的 aiomysql 连接生命周期，ORM 版本展示查询函数的接口形态（基础设施在第 13 章落地）。

# native_sql.py
import os

import aiomysql


async def find_book_by_isbn(isbn: str) -> dict | None:
    conn = await aiomysql.connect(
        host=os.getenv("DB_HOST", "127.0.0.1"),
        port=int(os.getenv("DB_PORT", "3306")),
        user=os.getenv("DB_USER", "root"),
        password=os.getenv("DB_PASSWORD", ""),
        db=os.getenv("DB_NAME", "blog"),
    )
    try:
        async with conn.cursor() as cur:
            await cur.execute(
                "SELECT id, title, isbn FROM books WHERE isbn = %s",
                (isbn,),
            )
            row = await cur.fetchone()
            if row is None:
                return None
            return {"id": row[0], "title": row[1], "isbn": row[2]}
    finally:
        conn.close()

原生版本中的 %s 是 aiomysql 的参数占位符，(isbn,) 作为独立参数传入——绝不能用 f-string 拼进去。try/finally 保证无论查询成功、未命中还是抛异常，连接都会被关闭。在高并发场景下，每次查询都新建连接是不可接受的——通常还需要引入连接池，代码量进一步膨胀。

# orm_style.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.book import Book


async def find_book_by_isbn(db: AsyncSession, isbn: str) -> Book | None:
    stmt = select(Book).where(Book.isbn == isbn)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

ORM 版本把会话作为参数传入。函数不创建也不关闭会话——创建由依赖注入的 get_async_db 负责（第 13 章），关闭由 async with 上下文自动完成。scalar_one_or_none() 明确要求结果至多一条——如果 ISBN 唯一约束意外失效导致多行，它会抛异常而不是静默取第一条。await db.execute 把协程交还给事件循环，数据库 I/O 期间服务器可以处理其他请求。

避坑指南

关联对象的懒加载是 N+1 的根源。先查 N 本书，再为每本书各查一次作者——ORM 的 lazy loading 让这种模式太容易写出。应根据访问模式在查询时就用 selectinload 或 joinedload 预加载关联，而不是等序列化阶段触发隐式查询。
Session 关闭后不要访问 ORM 对象的属性。对象离开原会话后处于 detached 状态，访问未加载的关系或已过期属性可能触发 DetachedInstanceError。返回响应前在有效事务边界内完成所有字段加载。
SQL echo 适合开发，不适合生产。echo=True 会把每条生成的 SQL 打印到控制台，开发和测试时很有用。但生产环境长期开启会增加日志 I/O，还可能暴露参数信息。线上应该改用慢查询监控和 APM 工具。
await db.execute() 是关键 I/O 边界。无论是原生 await cur.execute() 还是 ORM 的 await db.execute()，都在等待数据库响应时协作式让出事件循环。区别在于：在原生路径中，你还要手动管理连接和游标的生命周期；在 ORM 路径中，这些由会话自动处理。

面试 QA

Q1 [原理]: ORM 与原生 SQL 的核心差异是什么，哪些场景更适合原生 SQL？

ORM 以对象和查询表达式为入口，承担 SQL 生成和结果映射；原生 SQL 以参数化语句为入口，保留对执行细节的直接控制。

对于常规 CRUD——字段和对象属性对应稳定——ORM 能以更少的代码完成查询，同时集中管理会话和事务。但复杂报表、批量写入、或需要精确控制执行计划（FORCE INDEX、STRAIGHT_JOIN）时，原生 SQL 往往更直接。关键原则是：即使使用原生 SQL，也必须用参数绑定而不是字符串拼接。

落到图书 API：普通书目列表和详情查询适合 ORM；抢占唯一书目库存或批量同步外部 ISBN 数据时，可评估 MySQL 专有语句（如 INSERT ... ON DUPLICATE KEY UPDATE），并通过压测和 EXPLAIN 验证选择，而不是仅凭代码长短决定。

Q2 [项目]: 引入异步 ORM 后，图书 API 的数据访问层应该如何组织？

按三层组织：模型声明（app/models/）负责表、列、约束和关系，不承载 HTTP 参数；事务控制由异步 yield 依赖托管——请求开始时产出 Session，结束前显式 commit 或 rollback，响应后自动关闭；查询封装（app/dao/）接收调用方传入的 Session 并执行可复用查询。

跨多个 DAO 的复杂事务（如借书 = 改 Book.status + 新增 Borrow 行）由服务层协调。路由层只处理 HTTP 输入输出，不感知 Session 的创建和销毁。这种分层让单元测试可以在不启动 FastAPI 的情况下调用 DAO 验证业务逻辑，让集成测试可以在不接触业务代码的情况下验证 HTTP 契约。

小结

本篇建立了 ORM 的对象映射概念，并把 aiomysql 原生 SQL 和 SQLAlchemy 异步 ORM 两条路径并排对比——原生路径强调显式管理连接、游标和结果转换，ORM 路径强调会话边界和映射层协作。两者没有绝对的优劣，选择取决于你对"控制权 vs 便利性"的偏好，以及当前场景是"常规 CRUD"还是"需要精确控制执行计划"。

下一篇《13 SQLAlchemy ORM》将把本章的概念落地为真实的异步基础设施：create_async_engine、async_sessionmaker、AsyncSession 和 get_async_db——这四件套是整个数据层的骨架。

FastAPI系列-12-ORM介绍