SQLAlchemy 2.0 核心概念与异步支持实战

搞了这么多年 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_keyTrue) name: Mapped[str] mapped_column(String(50)) age: Mapped[int] mapped_column(default0)带上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_keyTrue) 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()取唯一实体或 Nonesession.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_commitFalse。我说说为什么要这么改。1.x 时代默认每次commit()之后把会话里所有对象标记为过期下一次访问属性时发 SQL 去数据库重新加载。在同步代码里这个行为问题不大但到了异步环境session.commit()之后如果你在异步代码里访问user.nameORM 会尝试同步地刷新属性结果就是经典的MissingGreenlet错误。把expire_on_commitFalse设上提交后对象保留当前状态不触发额外查询可以规避一大部分奇怪的异步报错。事务边界的写法我比较推荐session.begin()的上下文管理器形式with session_factory() as session, session.begin(): session.add(user1) session.add(user2) # 这里任何异常都会自动回滚这段代码等价于手动commitrollback但可读性好很多。异步场景对应的是async with AsyncSessionLocal() as session, session.begin(): session.add(user1) await session.flush()注意session.begin()块内如果涉及主键生成可能需要await session.flush()把新增对象刷到数据库否则后续关联查询可能拿不到自增 id。另外2.0 已经正式移除了autocommitTrue这个参数1.x 时代有人为了省事开启 autocommit现在官方明确不支持这种模式所有写操作都应在显式事务里进行。2.3 relationship 关系加载别让 N1 拖垮接口ORM 的关系映射本来是福利但用不好就是性能炸弹。最典型的是 N1查了 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_populatesuser, lazyselectin)我通常的做法是一对多且数据量可控的关系用lazyselectin很少访问的关系保持默认懒加载但查询入口里按需options(selectinload(...))涉及大量集合的场景可以考虑 2.0 新增的write_only关系避免 ORM 在内存里维护整个集合。关于write_only它是 2.0 用来替代lazydynamic的。旧版dynamic返回一个查询对象灵活但实现复杂write_only的思路是“这个关系只用来做写操作”不会在父对象上维护整个集合。订单量很大的场景声明orders: Mapped[list[Order]] relationship(lazywrite_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( postgresqlasyncpg://user:passlocalhost/dbname, echoTrue, pool_size5, max_overflow10, ) AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse)注意连接串里的驱动名必须换成对应异步驱动postgresqlasyncpg、mysqlaiomysql或asyncmy、sqliteaiosqlite。如果你用普通同步驱动写进异步引擎启动时不一定立刻报错但第一个查询执行时会告诉你驱动不支持。实际项目中我见过不少人试图在同一份代码里同时用同步和异步引擎。技术上是可以的数据库同一个但引擎不同。但千万别让同一个Session对象在同步和异步之间混用因为AsyncSession内部依赖 greenlet 调度一旦把它的方法当成同步调用就会报错。渐进迁移时比较稳妥的做法先保留同步引擎跑旧代码新模块用异步引擎两个引擎各管各的 session。异步引擎的连接池同样支持pool_size、max_overflow、pool_pre_ping等参数。以 PostgreSQL 为例如果使用默认max_connections100而应用是多个 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(namename, ageage) 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()如果不 flushawait 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()如果模型声明了lazyselectin默认就不会有这个问题。使用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或lazyselectin不要在业务代码里到处写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)因为如果查询里带了 joincount 的语义会变显式指定 FROM 可以让意图更清楚也能避免生成意料之外的 SQL。另一个迁移时必看的点是Session的autocommit。1.x 时代有些人靠sessionmaker(autocommitTrue)偷懒2.0 彻底移除这个参数迁过来第一步就是把所有依赖 autocommit 的代码改成显式事务。还有一点1.x 的lazydynamic在 2.0 里已经不再推荐建议改成lazywrite_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_commitFalse)设置好是迁移第一步。外键和类型注解冲突。某些模型如果字段类型是Optional[str]Mapped 注解应写成Mapped[Optional[str]]否则 SQLAlchemy 会把它推断成非空字段。我在某次迁移时就因为Mapped[str]漏了Optional建表把 email 字段设成了 NOT NULL插入数据一直报错。selectinload可能导致子查询过大。如果关系集合特别大比如几万条selectinload的IN (SELECT ...)子查询会有性能问题。此时通常改用joinedload或分页查询不要无脑 selectinload。5. 常见问题与排查速查表5.1 异步场景的典型报错报错信息原因处理方式MissingGreenlet异步代码中触发了懒加载加预加载选项或使用AsyncAttrs或lazyselectingreenlet_spawn has not been called直接调用了异步 session 的同步方法检查是否漏写awaitTypeError: Object ... is not a coroutine忘了 await给相应调用加awaitAttributeError: AsyncSession object has no attribute queryAsyncSession没有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_connections100为例如果应用起了 4 个 worker每个 worker 的 pool_size10、max_overflow5那理论上最多会占用 4 * 15 60 个连接还有余量给管理和后台任务。反过来如果数据库连接数限制很低就必须调小 pool_size同时设置合理的pool_timeout不然应用高峰期会出现获取连接超时。还有一个小技巧开发调试时给create_async_engine加echoTrueSQLAlchemy 会把执行的所有 SQL 打到终端。我排查 N1 问题最快的方式就是开着 echo看到底发了多少条查询。线上环境别开刷日志太猛。5.3 老项目升级后的渐进验证最后分享一下我在模拟项目迁移里的收尾做法把所有模块的 SQL 执行日志收集起来跑一遍回归测试对比升级前后同一条业务路径的 SQL 数量。凡是数量明显变多的大概率是懒加载策略变了或遗漏了预加载。这个对比不一定要自动化人工看 echo 日志也能发现大部分问题。我在实际项目中体会到迁移 2.0 最花时间的往往不是查 API 文档而是把query链式习惯从脑子里清出去。每当你条件反射写出session.query(...)就提醒自己停下来换成selectexecute。坚持一两个星期新写法就会变成肌肉记忆。至于异步等你把同步方式理顺了AsyncSession其实只是多写几个await的事。