搞了这么多年 Python ORM，我越来越觉得 SQLAlchemy 2.0 是一套全新的心智模型，而不是 1.x 的小修小补。最被低估的其实是它把查询入口、映射声明、事务管理、加载策略全部收敛到同一套 API 上，异步支持（AsyncSession/AsyncEngine）只是这套新模型水到渠成的结果。也就是说，你如果不理解select()+session.execute()这条主链路，就算照抄异步代码，也会在事务边界和懒加载上踩出一片坑。
这篇文章围绕“SQLAlchemy 2.0 核心概念与异步支持”展开，我会把 2.0 的查询模型、声明式映射、事务边界、关系加载、AsyncSession 实战逐个拆开，再给一份从 1.x 迁移到 2.0 的对照清单。适合正在做技术选型的开发者、准备升级老项目的人，以及已经在异步项目里被 MissingGreenlet 折磨过的朋友。
1. 2.0 到底改了什么东西
1.1 查询模型统一：Query API 退居二线
1.x 时代我们最熟悉的写法是session.query(User).filter(...).all()，这套链式 API 用起来确实顺手，但问题在于Query是 ORM 层专用的查询对象，而 Core 层的select()又是另一套逻辑，两套 API 各自为政，导致新手经常搞不清楚“什么时候用 Query，什么时候用 select”。
2.0 把这层窗户纸捅破了：不管你是直接操作表，还是操作 ORM 映射实体，统一走select()构造语句，再用session.execute(stmt)执行。Query对象在 2.0 里虽然还能用，但已经被标记为 legacy 风格，官方文档推荐新代码一律使用 2.0 风格。
比如下面这段 2.0 风格代码：
from sqlalchemy import select stmt = ( select(User) .where(User.age >= 18) .order_by(User.age.desc()) .limit(10) ) users = session.execute(stmt).scalars().all()把filter()换成where()，把query(User)换成select(User)，看起来只是命名变化，但本质区别是Select对象可以直接传给 Core 的执行接口，也可以继续叠加 join、子查询等条件。更重要的是，Select是不可变对象，每次.where()都会返回一个新对象，可以放心复用基础语句。
我实际项目里常这么玩：把常用的过滤条件抽成函数，返回一个基础的Select，然后各接口在此基础上继续加条件。
def base_user_stmt(): return select(User).where(User.is_active == True) stmt = base_user_stmt().where(User.role == "admin")这种写法在 1.x 里不是做不到，但Query是可变对象，复用起来容易互相污染；而Select每次叠加条件都返回新实例，源码层面就杜绝了流程之间互相影响的问题。
结果对象的处理也要重新适应。以前query(...).all()直接给你实体列表，现在session.execute()给你的是一个Result包装，你得告诉它你想怎么拆：
result.scalars()：把每一行里的第一个元素作为实体取出来，适合select(User)这种查询，配合.all()拿到实体列表；result.scalar()：取单个标量值，适合select(func.count())；result.mappings()：按列名映射成字典，适合查特定列；result.scalar_one()/scalar_one_or_none()：取唯一实体，如果结果不是恰好一条就报错或返回None。
很多新手抱怨“为什么 execute 之后还要调一次 scalars”，原因就是execute()并不知道你要实体还是行数据，它只负责执行并暴露一个通用的结果接口。这也是 2.0 把 ORM 和 Core 统一起来以后的必然结果，理解这一点之后，结果对象的各种方法就不会再搞混了。
1.2 声明式映射：Mapped 与 mapped_column 的正确用法
SQLAlchemy 2.0 在声明模型这边最大的变化是Mapped类型注解和mapped_column()的组合：
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import String, ForeignKey class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50)) age: Mapped[int] = mapped_column(default=0)带上Mapped[str]注解之后，IDE 的补全和类型检查会直接生效。我在某项目里见过十几处user.name.lower()这种调用，用了新写法之后编辑器立刻能提示name是str，很多隐患在保存代码的时候就暴露了，不需要等运行期再排查。
关于mapped_column()，一个常见疑问是“那Column还能用吗”。能用，2.0 仍然兼容老的Column写法，但如果你在同一个模型里混用Mapped注解和纯Column，有些边界情况会让 IDE 推断失效。我的建议是：新项目全部用mapped_column()，老项目迁移时按模型逐个统一改。
另外，外键和关系的声明顺序也值得注意。假如有个订单表要引用用户表：
class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))如果类定义里User还没出现，也可以直接用字符串"User"作为关系目标，2.0 的声明式解析器会在映射阶段处理。整体感受是，声明式映射现在更像“为普通 Python 类补充类型标注”，而不是“写表结构说明书”。
2. 新 API 的日常操作套路
2.1 select() 与 Result 的取数姿势
日常 CRUD 里，最常踩的坑是分不清取实体的姿势。我先给一张速查表：
| 需求 | 推荐写法 |
|---|---|
| 取所有实体 | session.execute(select(User)).scalars().all() |
| 取第一个实体 | session.execute(select(User)).scalars().first() |
| 取唯一实体 | session.execute(select(User).where(User.id == 1)).scalar_one() |
| 取唯一实体或 None | session.execute(...).scalar_one_or_none() |
| 取单个聚合值 | session.execute(select(func.count(User.id))).scalar() |
| 取多列字典 | session.execute(select(User.name, User.age)).mappings().all() |
注意scalar_one()在结果多于一条时会抛MultipleResultsFound，这不是 bug，是保护。我曾经在一段接口逻辑里用scalar_one()查用户，本意是只应该存在一个，结果查出两条才发现底层数据有脏数据。那个问题要是用.first()会静默吞掉，后面排错成本更大。
2.0 还有一个很实用的能力：Session.get()替代了原来的Query.get()，按主键取对象：
user = await session.get(User, 1)如果只是主键查询，session.get()比select().where(...)更直接，而且它会利用当前会话的 identity map 做缓存，同一个对象在同一个 session 里不会重复加载两份。对性能和对象一致性都有好处。
2.2 Session 生命周期与事务边界
Session 是 ORM 的工作单元，负责跟踪加载过的实体、新增的对象，以及它们的变化。在 2.0 里，最重要的一个配置变化是建议expire_on_commit=False。
我说说为什么要这么改。1.x 时代默认每次commit()之后把会话里所有对象标记为过期，下一次访问属性时发 SQL 去数据库重新加载。在同步代码里这个行为问题不大，但到了异步环境，session.commit()之后如果你在异步代码里访问user.name，ORM 会尝试同步地刷新属性，结果就是经典的MissingGreenlet错误。把expire_on_commit=False设上，提交后对象保留当前状态，不触发额外查询，可以规避一大部分奇怪的异步报错。
事务边界的写法，我比较推荐session.begin()的上下文管理器形式：
with session_factory() as session, session.begin(): session.add(user1) session.add(user2) # 这里任何异常都会自动回滚这段代码等价于手动commit+rollback，但可读性好很多。异步场景对应的是：
async with AsyncSessionLocal() as session, session.begin(): session.add(user1) await session.flush()注意session.begin()块内如果涉及主键生成，可能需要await session.flush()把新增对象刷到数据库，否则后续关联查询可能拿不到自增 id。另外，2.0 已经正式移除了autocommit=True这个参数，1.x 时代有人为了省事开启 autocommit，现在官方明确不支持这种模式，所有写操作都应在显式事务里进行。
2.3 relationship 关系加载：别让 N+1 拖垮接口
ORM 的关系映射本来是福利，但用不好就是性能炸弹。最典型的是 N+1：查了 100 个用户，然后循环里访问user.orders，每次访问都发一条 SQL，一共 101 条查询，接口直接卡死。
2.0 推荐两种解决思路。
第一种是查询时使用selectinload或joinedload显式加载：
from sqlalchemy.orm import selectinload stmt = select(User).options(selectinload(User.orders)) users = session.execute(stmt).scalars().all() for user in users: print(len(user.orders)) # 不会触发新的 SQL第二种是把relationship的lazy设为"selectin"，这样默认加载关系时使用IN子查询批量拉取：
class User(Base): __tablename__ = "users" orders: Mapped[list["Order"]] = relationship(back_populates="user", lazy="selectin")我通常的做法是：一对多且数据量可控的关系用lazy="selectin"；很少访问的关系保持默认懒加载，但查询入口里按需options(selectinload(...))；涉及大量集合的场景可以考虑 2.0 新增的write_only关系，避免 ORM 在内存里维护整个集合。
关于write_only，它是 2.0 用来替代lazy="dynamic"的。旧版dynamic返回一个查询对象，灵活但实现复杂；write_only的思路是“这个关系只用来做写操作”，不会在父对象上维护整个集合。订单量很大的场景，声明orders: Mapped[list["Order"]] = relationship(lazy="write_only")后，你可以user.orders.add(order)添加关联，但不用担心访问user.orders时把成千上万的订单全拉进内存。查询还是走select(Order).where(Order.user_id == user.id)。这个特性特别适合订单、日志这类只追加、量大、不需要常驻内存的集合。
3. 异步支持：AsyncSession 落地实战
3.1 环境准备与异步引擎创建
异步支持是 SQLAlchemy 2.0 的重头戏。第一件事就是创建异步引擎：
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker engine = create_async_engine( "postgresql+asyncpg://user:pass@localhost/dbname", echo=True, pool_size=5, max_overflow=10, ) AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)注意连接串里的驱动名必须换成对应异步驱动：postgresql+asyncpg、mysql+aiomysql（或asyncmy）、sqlite+aiosqlite。如果你用普通同步驱动写进异步引擎，启动时不一定立刻报错，但第一个查询执行时会告诉你驱动不支持。
实际项目中我见过不少人试图在同一份代码里同时用同步和异步引擎。技术上是可以的：数据库同一个，但引擎不同。但千万别让同一个Session对象在同步和异步之间混用，因为AsyncSession内部依赖 greenlet 调度，一旦把它的方法当成同步调用就会报错。渐进迁移时比较稳妥的做法：先保留同步引擎跑旧代码，新模块用异步引擎，两个引擎各管各的 session。
异步引擎的连接池同样支持pool_size、max_overflow、pool_pre_ping等参数。以 PostgreSQL 为例，如果使用默认max_connections=100，而应用是多个 worker 模式的 Web 服务，连接池设置太大会导致数据库连接被占满，所以 pool_size 要根据 worker 数和数据库上限来估算。
3.2 异步 CRUD 与事务写法
一个完整的异步 CRUD 代码块：
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select async def create_user(session: AsyncSession, name: str, age: int): user = User(name=name, age=age) session.add(user) await session.commit() return user async def fetch_user(session: AsyncSession, user_id: int): return await session.get(User, user_id) async def update_age(session: AsyncSession, user_id: int, new_age: int): user = await session.get(User, user_id) if user: user.age = new_age await session.commit() async def delete_user(session: AsyncSession, user_id: int): user = await session.get(User, user_id) if user: await session.delete(user) await session.commit()这段代码里值得注意的点：
- 新增对象后，如果你需要在事务内立刻使用数据库生成的 id，可以手动
await session.flush()；如果不 flush，await session.commit()时也会自动 flush，所以 commit 完成后user.id已经可用了。真正不行的是把所有操作放在一个事务外，commit 后再去依赖某些需要 reload 才能拿到的列值，那才需要await session.refresh(user)。 - 异步 session 的
get、execute、commit、delete、refresh都是协程，必须await。少写一个await的报错方式不是“返回值不对”，而是一堆coroutine object is not callable或TypeError，排查起来反而更麻烦。
批量场景下，2.0 也支持通过session.execute(insert(...).returning(...))直接拿自增 id。比如批量插入订单：
from sqlalchemy import insert stmt = insert(Order).values( [{"user_id": 1, "amount": 100}, {"user_id": 2, "amount": 200}] ).returning(Order.id) ids = (await session.execute(stmt)).scalars().all()这个写法在异步场景下性能会好很多，因为它只发一条 SQL，而不是逐条 insert。
3.3 MissingGreenlet 和 AsyncAttrs
这一节必须重点讲 MissingGreenlet。这个错误信息在 SQLAlchemy 异步社区几乎等于“你触发了懒加载”。
简单解释一下：AsyncSession在底层用 greenlet 把“同步的数据库驱动调用”包装成可暂停的协程，但只有在外层 await 调用链上才有 greenlet 环境。当你从异步代码里访问一个尚未加载的 relationship 属性时，ORM 想同步去执行 SQL，却发现当前没有 greenlet 上下文，于是抛MissingGreenlet。
我之前就被这个坑过：查询用户列表后，想顺手统计每个用户的订单数：
users = await session.execute(select(User)).scalars().all() for user in users: print(len(user.orders)) # MissingGreenlet解决方式有几种：
- 查询时加上加载选项：
stmt = select(User).options(selectinload(User.orders)) users = await session.execute(stmt).scalars().all()如果模型声明了
lazy="selectin"，默认就不会有这个问题。使用
AsyncAttrs显式异步加载字段：
from sqlalchemy.ext.asyncio import AsyncAttrs class Base(DeclarativeBase): pass class User(Base, AsyncAttrs): ... orders: Mapped[list["Order"]] = relationship(...)然后访问时：
orders = await user.awaitable_attrs.orders注意是await user.awaitable_attrs.orders，不是await user.orders。awaitable_attrs是一个代理对象，它知道当前是异步环境，会以异步方式触发懒加载。这一点对老模型迁移特别友好，因为不需要把每个查询都改成预加载。
我的实际建议是：新项目优先用selectinload或lazy="selectin"，不要在业务代码里到处写awaitable_attrs——它好用，但本质是给“偶尔访问”的场景准备的，高频访问最好还是预加载。
4. 从 1.x 迁移到 2.0 的实操清单
4.1 按功能模块替换 API
迁移不要一次性大改，我建议按功能模块拆开做。最核心的替换映射表：
| 1.x 写法 | 2.0 推荐写法 |
|---|---|
session.query(User) | select(User) |
.filter(...) | .where(...) |
.filter_by(...) | .where(User.xxx == ...) |
.get(id) | session.get(User, id) |
.all() | session.execute(stmt).scalars().all() |
.first() | session.execute(stmt).scalars().first() |
.one() | session.execute(stmt).scalar_one() |
.count() | select(func.count()).select_from(User) |
有个容易被忽略的坑：session.query(User).count()在 2.0 里没有直接等价的一行写法，通常用select(func.count()).select_from(User).where(...)。推荐显式写select_from(User)，因为如果查询里带了 join，count 的语义会变，显式指定 FROM 可以让意图更清楚，也能避免生成意料之外的 SQL。
另一个迁移时必看的点是Session的autocommit。1.x 时代有些人靠sessionmaker(autocommit=True)偷懒，2.0 彻底移除这个参数，迁过来第一步就是把所有依赖 autocommit 的代码改成显式事务。
还有一点，1.x 的lazy="dynamic"在 2.0 里已经不再推荐，建议改成lazy="write_only"。如果业务逻辑依赖user.orders.filter(...)这种动态查询方式，改write_only后写法会变，需要额外适配查询部分。
4.2 迁移时的常见坑点
结果对象变了。老代码
for user in session.query(User)是直接的实体迭代；新代码你写成for row in session.execute(select(User))拿到的每个元素是Row，要row[0]或者改写成scalars()。这个语法层面的变化，迁移量最大的反而是直观的循环体。expire_on_commit会影响 session 复用。老代码沿用默认True时，commit 后再次访问实体属性会重新发 SQL，这在异步环境特别容易炸。把sessionmaker(expire_on_commit=False)设置好是迁移第一步。外键和类型注解冲突。某些模型如果字段类型是
Optional[str]，Mapped 注解应写成Mapped[Optional[str]]，否则 SQLAlchemy 会把它推断成非空字段。我在某次迁移时就因为Mapped[str]漏了Optional，建表把 email 字段设成了 NOT NULL，插入数据一直报错。selectinload可能导致子查询过大。如果关系集合特别大（比如几万条），selectinload的IN (SELECT ...)子查询会有性能问题。此时通常改用joinedload或分页查询，不要无脑 selectinload。
5. 常见问题与排查速查表
5.1 异步场景的典型报错
| 报错信息 | 原因 | 处理方式 |
|---|---|---|
MissingGreenlet | 异步代码中触发了懒加载 | 加预加载选项，或使用AsyncAttrs，或lazy="selectin" |
greenlet_spawn has not been called | 直接调用了异步 session 的同步方法 | 检查是否漏写await |
TypeError: Object ... is not a coroutine | 忘了 await | 给相应调用加await |
AttributeError: 'AsyncSession' object has no attribute 'query' | AsyncSession没有query方法 | 用select()+execute() |
这里要提醒一下，AsyncSession和Session在 API 上不是完全一致的，最明显的就是AsyncSession没有.query()方法。所以如果你的代码还保留session.query(User)，切到异步后第一步就是改成 2.0 风格，这也是为什么前面一直强调“先理解 select()+execute() 模型，再谈异步”的原因。
5.2 事务与连接池的实战建议
事务边界方面，最忌讳的是在async with session.begin()块里执行耗时太久的任务。事务长时间不提交会一直占用数据库连接，在连接池较小或多服务共享数据库的情况下，很容易把连接耗尽。我一般在事务块里只放 ORM 操作和必要的缓存读取，业务计算尽量放在事务外。
连接池参数也是一门学问。pool_size对应常驻连接数，max_overflow是峰值时额外创建的上限。以 PostgreSQL 默认max_connections=100为例，如果应用起了 4 个 worker，每个 worker 的 pool_size=10、max_overflow=5，那理论上最多会占用 4 * 15 = 60 个连接，还有余量给管理和后台任务。反过来，如果数据库连接数限制很低，就必须调小 pool_size，同时设置合理的pool_timeout，不然应用高峰期会出现获取连接超时。
还有一个小技巧：开发调试时给create_async_engine加echo=True，SQLAlchemy 会把执行的所有 SQL 打到终端。我排查 N+1 问题最快的方式就是开着 echo，看到底发了多少条查询。线上环境别开，刷日志太猛。
5.3 老项目升级后的渐进验证
最后分享一下我在模拟项目迁移里的收尾做法：把所有模块的 SQL 执行日志收集起来，跑一遍回归测试，对比升级前后同一条业务路径的 SQL 数量。凡是数量明显变多的，大概率是懒加载策略变了或遗漏了预加载。这个对比不一定要自动化，人工看 echo 日志也能发现大部分问题。
我在实际项目中体会到，迁移 2.0 最花时间的往往不是查 API 文档，而是把query链式习惯从脑子里清出去。每当你条件反射写出session.query(...)，就提醒自己停下来换成select+execute。坚持一两个星期，新写法就会变成肌肉记忆。至于异步，等你把同步方式理顺了，AsyncSession其实只是多写几个await的事。
