第1章：SQLAlchemy 术语全景与分层架构原理

davidwang456

322人浏览 · 2026-08-14 13:43:35

davidwang456 · 2026-08-14 13:43:35 发布

一、项目背景

"这个月的账务对不上！"财务总监把报表摔在桌子上。

星云电商的订单中台已经撑过了三个大型促销活动，但技术债也像雪球一样越滚越大。最初为了快，团队直接用裸 DBAPI（psycopg2/asyncpg）写 SQL 字符串拼接。随着业务复杂度指数级增长，代码库里已经充斥着 2000+ 行的手写拼接 SQL、散落各处的连接管理逻辑、以及"先查后改再写"的零散事务代码。

上一季度出现了三次 P0 事故：

一次 SQL 注入漏洞导致用户数据泄漏——开发人员把用户输入直接拼接进 WHERE 子句。
连接泄漏事故——某次促销高峰，数据库连接数被打满，新请求全部超时，因为代码里有一处 conn.execute() 后忘了 conn.close()。
事务脏读——下单接口由于没有正确设置隔离级别，用户在极短时间内重复提交，库存被扣到负数。

架构团队痛定思痛，决定引入 ORM 框架来规范化数据访问层。经过横向对比 Django ORM、Peewee、PonyORM 和 SQLAlchemy，最终选定了 SQLAlchemy 2.0——它既提供了高层的 ORM 抽象，又保留了低层的 Core 直接操作 SQL 的能力，而且 2.0 版本统一了 API 风格，彻底告别了 1.x 时代的 Query 与 Session.execute 两套写法并存的问题。

然而，团队面临的第一个障碍是术语混乱：Engine、Connection、Session、MetaData、Mapper、Identity Map、Unit of Work……每个人对这些概念的理解都停留在"大概知道"，沟通成本极高。本章作为专栏开篇，就是要建立统一术语词典，让团队对 SQLAlchemy 的分层架构形成共识。

二、项目设计

场景：周一下午，星云订单中台团队的技术评审会上，小胖正抱着一袋薯片，看着架构师大师在白板上画了满满一版图。

小胖（嚼着薯片）：“大师，这满满一白板是啥啊？Engine、Connection、Session……这不就是个数据库连接吗？我们用 psycopg2 的时候，一个 connect() 就搞定的事儿，为啥 SQLAlchemy 要搞这么多层？这不就跟食堂打饭一样——一个窗口解决的问题，非要分成选菜区、结账区、取餐区？”

大师（放下白板笔）：“小胖这个比喻很好。食堂为什么要分区？因为当只有 10 个人吃饭时，一个窗口没问题。但当 500 个人同时涌入，你需要有人专门管窗口调度、有人管菜、有人管结账。SQLAlchemy 的分层也是同样的道理。”

小白（推了推眼镜）：“但我觉得需要警惕过度抽象。我就想问：如果我只想执行一条 SELECT * FROM users，到底要走多少层代码？每一层的职责边界是什么？有没有可能出现’为了分层而分层’造成的性能开销？”

大师：“好问题。我们一层层拆解。”

（大师在白板上重新画了一个五层架构图）

应用代码
   │
   ▼
┌──────────────────────────────────────┐
│  ORM (Session, Mapper, UnitOfWork)   │  ← 对象关系映射，管理 Python 对象与行的转换
├──────────────────────────────────────┤
│  Core Expression (select/insert/...) │  ← SQL 抽象层，用 Python 对象拼装 SQL
│  Core Schema    (MetaData/Table/Col) │  ← 表结构元数据描述
├──────────────────────────────────────┤
│  Engine / Pool / Connection          │  ← 连接管理与执行入口
├──────────────────────────────────────┤
│  Dialect (PostgreSQL / MySQL / ...)   │  ← 方言：将通用 AST 转为特定数据库的 SQL
├──────────────────────────────────────┤
│  DBAPI (psycopg3 / asyncpg / ...)    │  ← 底层数据库驱动
├──────────────────────────────────────┤
│  数据库 (PostgreSQL / MySQL / ...)   │
└──────────────────────────────────────┘

大师：“最底层是 DBAPI，就是你们之前用的 psycopg2。往上是 Dialect，它负责一件事：把 SQLAlchemy 的通用表达式翻译成特定数据库的 SQL 方言。比如 LIMIT 10 在 MySQL 是 LIMIT 10，在 SQL Server 是 SELECT TOP 10。”

小胖：“哦——所以 Dialect 就是个翻译官？跟去广东喝茶，普通话说’来壶普洱’，翻译成粤语给服务员听？”

大师：“技术映射：Dialect = SQL 方言翻译层。对，就是翻译官。然后往上是 Engine 和 Connection。Engine 是一个不可变的工厂对象，它保存了数据库 URL、连接池配置、Dialect 实例等全局配置。而 Connection 是轻量级的——它代表一次真实的数据库连接。”

小白：“等等，你说 Engine 是工厂，Connection 是执行单元。那 Session 呢？我查文档的时候经常看到这两个概念混在一起。”

大师：“Session 是 ORM 层的概念，它不是连接本身，而是持有一个 Connection。Session 负责对象状态管理——你 session.add(user) 后，这个 user 对象进入了一个叫 Identity Map 的数据结构，Session 跟踪它的变更，在 flush 时把脏数据转为 SQL 并提交到 Connection。”

小胖（放下薯片）：“Identity Map？听起来像身份证登记处？”

大师：“技术映射：Identity Map = 主键索引缓存。完全正确。比如你两次从库里查 id=1 的用户，Session 只会发一次 SQL，第二次直接从 Identity Map 返回同一个 Python 对象。这保证了’在同一个 Session 生命周期内，同一主键对应同一个 Python 对象’。”

小白：“那 Unit of Work 又是什么？听名字像银行的对账处？”

大师：“技术映射：Unit of Work = 变更排序调度器。Unit of Work 是 flush 的核心。当你 session.flush() 时，它把 pending 状态的对象按外键依赖排序：先插主表再插子表，先删子表再删主表。然后发出一系列 INSERT/UPDATE/DELETE 语句。”

小胖：“那我能不能直接跳过 ORM，只用 Core？就像我只去食堂打饭不要套餐？”

大师：“这正是 SQLAlchemy 的设计精妙之处。它不是一个’全有或全无’的 ORM，而是分层的。如果你只需要执行简单 SQL，直接用 connection.execute(text('SELECT ...'))。如果你想用 Python 拼装 SQL，用 Core Expression：conn.execute(select(users).where(users.c.name == '张三'))。如果你需要对象关系映射，才上 ORM。”

小白：“还有一个关键问题：doc 里提到 2.0 使用的是统一的 session.execute(select(...)) 风格，而不是 1.x 的 session.query(User).filter(...)。这个改动的意义是什么？”

大师：“1.x 时代 ORM 有自己独立的查询构造体系（Query），和 Core 的 select() 是两套平行的 API。这意味着同样的过滤、排序逻辑，你要写两套不同的代码。2.0 统一后，select() 既可以在 Core 层用 conn.execute() 执行，也可以在 ORM 层用 session.execute() 执行——它们的查询构造方式完全一致，区别仅在于返回结果：Core 返回 Row，ORM 返回 ORM 对象。”

小胖：“懂了！就像食堂的菜，同一个厨师做的，你在窗口吃就是堂食（Core），端回座位吃就是外带（ORM）——但菜是一样的！”

大师：“技术映射：2.0 统一 API = Core/ORM 共用同一查询构造器。总结一下：Engine 管’怎么连’，Connection 管’怎么执行’，Dialect 管’怎么翻译’，MetaData 管’表长啥样’，Session 管’对象怎么变’，Identity Map 管’谁是谁’，Unit of Work 管’变更怎么排’。”

小胖：“20 行代码走通全链路？真的假的？”

大师：“来，我们自己动手。”

三、项目实战

实战目标

用不超过 20 行代码走通 Engine → Connection → Result 的完整路径，验证 SQLAlchemy 的安装和基础链路。

环境准备

# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 安装 SQLAlchemy 2.0 及驱动
pip install sqlalchemy==2.0.36 psycopg2-binary

注意：本章仅用最小依赖验证核心链路。完整的 Docker + PostgreSQL 环境在第 2 章搭建。

步骤一：创建 Engine 并执行第一条 SQL

目标：用 create_engine 创建引擎，用 engine.connect() 获取连接，执行 SELECT 1。

from sqlalchemy import create_engine, text

# 1. 创建 Engine（不可变工厂对象，包含连接池和 Dialect）
# URL 格式: dialect+driver://user:password@host:port/database
engine = create_engine(
    "postgresql+psycopg2://user:pass@localhost:5432/mydb",
    echo=True,   # 打印每条 SQL，方便观察
    pool_size=5, # 连接池基础大小
)

# 2. 获取 Connection 并执行
with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))
    row = result.fetchone()
    print(f"数据库返回: {row}")  # 输出: 数据库返回: (1,)
    conn.commit()

运行结果：

2026-08-14 10:00:00 INFO sqlalchemy.engine.Engine SELECT 1
2026-08-14 10:00:00 INFO sqlalchemy.engine.Engine [generated in 0.0005s] {}
数据库返回: (1,)
2026-08-14 10:00:00 INFO sqlalchemy.engine.Engine COMMIT

关键观察：

echo=True 输出了 SQL 语句和执行时间，这在开发和排查问题时极其有用。
with engine.connect() 自动管理连接的生命周期：进入时 checkout 一个连接，退出时 commit/rollback 并归还连接池。
text("SELECT 1") 是 SQLAlchemy 的文本 SQL 构造，用于在不使用 ORM/Core Expression 时直接写原生 SQL。

步骤二：用 Core Expression 执行查询

目标：使用 select() 构造 SQL，观察编译结果。

from sqlalchemy import select, column, table, MetaData

# 手动构造 Core Table 元数据（第4章会详解）
metadata = MetaData()
users = table("users", column("id"), column("name"), column("age"))

# 构造 SELECT 查询
stmt = select(users).where(users.c.name == "张三").limit(10)

# 查看编译后的 SQL（不执行）
from sqlalchemy.dialects import postgresql
compiled = stmt.compile(dialect=postgresql.dialect())
print(f"编译后的 SQL: {compiled}")
print(f"绑定参数: {compiled.params}")

# 实际执行
with engine.connect() as conn:
    result = conn.execute(stmt)
    for row in result:
        print(row)

运行结果：

编译后的 SQL: SELECT users.id, users.name, users.age
FROM users
WHERE users.name = %(name_1)s
 LIMIT %(param_1)s
绑定参数: {'name_1': '张三', 'param_1': 10}

关键观察：

select(users) 返回的是一个 Select 对象，不是字符串。它可以被编译、修改、复用。
参数自动绑定为 %(name_1)s，这就是 SQLAlchemy 的参数化查询——值永远不直接拼进 SQL 字符串中，从根本上杜绝了 SQL 注入。

步骤三：用 ORM 声明模型并查询

目标：完成最小的 ORM 映射与查询链路。

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# 声明式基类
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    age: Mapped[int] = mapped_column(nullable=True)

# 创建表
Base.metadata.create_all(engine)

# ORM 查询
with Session(engine) as session:
    # 2.0 统一风格：用 select() 而不是 session.query()
    stmt = select(User).where(User.name == "张三")
    result = session.execute(stmt)
    user = result.scalars().first()
    print(f"查到的用户: {user}")

运行结果：

2024-01-01 10:00:01 INFO sqlalchemy.engine.Engine
CREATE TABLE users (
    id SERIAL NOT NULL,
    name VARCHAR NOT NULL,
    age INTEGER,
    PRIMARY KEY (id)
)
...
查到的用户: <User id=1, name='张三', age=25>

关键观察：

Base.metadata.create_all(engine) 根据 ORM 模型定义自动生成 DDL 并执行。
session.execute(select(User)) 是 2.0 统一 API 的核心——ORM 查询使用和 Core 完全相同的 select() 构造。
result.scalars() 从 Row 对象中提取出 ORM 实体，返回标量结果。

可能遇到的坑：

ModuleNotFoundError: No module named 'psycopg2'

原因：未安装 PostgreSQL 驱动。
解决：pip install psycopg2-binary。如果编译失败（Windows 常见），可改用 pip install psycopg2-binary。

OperationalError: could not connect to server

原因：PostgreSQL 未启动或连接信息有误。
解决：确认 PostgreSQL 运行中，检查 URL 中的 host/port/user/password/database。

ArgumentError: Textual SQL expression should be explicitly declared as text()

原因：在 conn.execute() 中直接传入了字符串而不是 text() 包裹的 SQL。
解决：写成 conn.execute(text("SELECT 1"))，而不是 conn.execute("SELECT 1")。这是 2.0 的安全要求，防止意外拼接不可信字符串。

TypeError: 'User' object is not iterable

原因：用了 result.scalar() 而不是 result.scalars().first()（注意复数形式）。
解决：scalars() 是返回迭代器的方法，scalar() 是获取单个标量值的方法。

完整代码清单

"""ch01_architecture_walkthrough.py —— 20 行代码走通 SQLAlchemy 全链路"""
from sqlalchemy import create_engine, text, select, table, column, MetaData
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.dialects import postgresql

# ============================
# 1. Core: Engine + Connection
# ============================
engine = create_engine(
    "postgresql+psycopg2://user:pass@localhost:5432/mydb",
    echo=True, pool_size=5
)

with engine.connect() as conn:
    row = conn.execute(text("SELECT 1")).fetchone()
    print(f"Core 查询: {row}")

# ============================
# 2. Core: Expression 编译
# ============================
metadata = MetaData()
t_users = table("users", column("id"), column("name"), column("age"))
stmt = select(t_users).where(t_users.c.name == "张三")
compiled = stmt.compile(dialect=postgresql.dialect())
print(f"SQL: {compiled}  |  参数: {compiled.params}")

# ============================
# 3. ORM: 声明映射 + 查询
# ============================
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    age: Mapped[int] = mapped_column(nullable=True)

Base.metadata.create_all(engine)

with Session(engine) as session:
    user = session.execute(select(User).where(User.name == "张三")).scalars().first()
    print(f"ORM 查询: {user}")

测试验证

# test_ch01_health.py
import pytest
from sqlalchemy import create_engine, text

def test_engine_health():
    """验证 Engine 能正常连接并执行简单查询"""
    engine = create_engine(
        "postgresql+psycopg2://user:pass@localhost:5432/mydb",
        echo=False
    )
    with engine.connect() as conn:
        result = conn.execute(text("SELECT 2 + 2 AS result"))
        row = result.fetchone()
        assert row[0] == 4

def test_text_sql_injection_safe():
    """验证参数化查询可防护 SQL 注入"""
    engine = create_engine("sqlite:///:memory:")
    with engine.connect() as conn:
        conn.execute(text("CREATE TABLE t (id INTEGER, name TEXT)"))
        conn.execute(text("INSERT INTO t VALUES (1, 'safe')"))
        conn.commit()
        # 模拟注入攻击
        malicious = "safe' OR '1'='1"
        result = conn.execute(
            text("SELECT * FROM t WHERE name = :name"),
            {"name": malicious}
        )
        assert result.fetchone() is None  # 参数化查询不会注入成功

运行测试：

pytest test_ch01_health.py -v

四、项目总结

优点与缺点

对比维度	裸 DBAPI（psycopg2）	SQLAlchemy 2.0
SQL 注入防护	手动 `%s` 占位符，容易遗漏	参数化查询默认执行，不可能拼接字符串
连接管理	手动 open/close，容易泄漏	Engine + Pool 自动管理，上下文管理器保证归还
代码复用	SQL 字符串散落各文件，重复度高	Expression 对象可组合复用
对象映射	手写 Row → Object 转换逻辑	ORM 自动映射，Identity Map 保证一致性
方言兼容	需要为不同数据库写不同 SQL	Dialect 层透明处理差异
学习成本	低，有 SQL 基础即可	较高，需理解多层抽象
性能开销	极小	ORM 层有少量抽象开销，Core 层可忽略

适用场景

推荐使用 SQLAlchemy 的场景：

业务逻辑复杂、模型关联多的中大型项目（如电商订单中台、ERP 系统）。
需要支持多种数据库的项目（Dialect 层屏蔽差异）。
团队已有统一的代码规范，需要 ORM 来约束数据访问模式。
需要数据库迁移管理的项目（配合 Alembic）。
需要灵活在 Core（高性能）和 ORM（开发效率）间切换的项目。

不推荐使用的场景：

极简的脚本/工具，只执行几条查询——直接用 sqlite3 模块更合理。
纯 ETL/数据管道类应用——SQLAlchemy 不是 ETL 工具，用 pandas + 原生驱动更合适。

注意事项

不要混用 1.x 和 2.0 API：如果在现有项目中看到 session.query(User).filter(...)，那是旧式写法。新代码统一使用 session.execute(select(User).where(...))。
echo=True 不要在生产环境开启：它会输出所有 SQL 到标准输出，性能影响显著且可能泄漏敏感数据。生产应使用结构化日志 + 采样。
Connection 用完及时归还：务必使用 with engine.connect() 或 with Session(engine) 上下文管理器，否则连接不会归还到连接池。

常见踩坑经验

案例 1：连接泄漏导致数据库连接数耗尽

现象：促销高峰期，新请求报 TimeoutError: QueuePool limit of size 5 overflow 10 reached。
根因：某处代码写了 conn = engine.connect()，但未关闭也未使用上下文管理器。
修复：改为 with engine.connect() as conn:，并加入连接池溢出告警。

案例 2：文本 SQL 忘记 text() 包装

现象：代码中 conn.execute("SELECT * FROM users WHERE id = " + user_id) 在测试环境正常，但在 SQLAlchemy 2.0 下直接报错。
根因：SQLAlchemy 2.0 强制要求原始 SQL 字符串必须用 text() 包装，防止开发者意外拼接不可信输入。
修复：conn.execute(text("SELECT * FROM users WHERE id = :id"), {"id": user_id})。

案例 3：create_all 在生产环境误操作

现象：运维在部署时误执行了 create_all，导致已有表被重建（数据丢失）。
根因：create_all 只检查表是否存在，不检查表结构是否匹配；如果表已存在则跳过。但如果删表操作介入则可能先删后建。
修复：生产环境永远用 Alembic 管理 Schema 变更（第14章详解），create_all 仅供本地开发使用。

思考题

SQLAlchemy 的核心分层设计中，Engine、Connection、Session 三者分别对应用餐场景中的哪些角色？如果需要在 Web 请求间共享一个 Engine 但每个请求使用独立的 Connection，应该如何设计？
某团队在使用 SQLAlchemy 1.x 的项目中有大量 session.query(User).filter(User.name == '张三').all() 的代码。现在要迁移到 2.0，请写出等价写法，并说明 session.execute(select(User)) 相比 session.query(User) 在查询复用上有何优势。