去年我接了一个数据服务重构的活儿存量代码全部是SQLAlchemy 1.4风格几十个模型文件、十几处手写 SQL 拼查询条件。咬着牙把整套东西迁到 2.0过程里踩了不少文档里没写明白的坑。这篇就当是自己的迁移笔记也把 2.0 最核心的变化和异步那套东西揉碎了讲讲。如果你正准备从 1.x 升级或者新项目想直接用 2.0 的异步特性可以按这篇的路径往下走。SQLAlchemy 2.0 说白了就是一次大统一把原来 ORM 和 Core 两套 API 的割裂感彻底抹平Session.execute()成了唯一的执行入口select()成了唯一的查询构造器同时把异步支持从“实验室”扶正成了官方一等公民。它不是什么新语言就是让 Python 生态里最老牌的 ORM 重新跟上了现代 Python 的类型注解和 asyncio 节奏。先说这篇适合谁被 1.4 的QueryAPI 折腾过、想搞清楚新版到底哪不一样的老人还有想直接上手异步数据库访问、但被文档里一堆AsyncSession、async_sessionmaker、greenlet术语劝退的新人。看完你至少能回答三个问题映射模型为什么这么写、Session和select()怎么配合、异步会话到底比同步快在哪、坑在哪。1. 2.0 版核心概念到底变在哪里很多人一上来就搜“SQLAlchemy 2.0 新特性”结果看到一堆Mapped、mapped_column、DeclarativeBase第一反应是“这不是换了个写法吗”。对表面是写法变了但骨子里是设计思路变了。1.1 两个 API 统一成了一个入口1.x 时代ORM 查询你要用session.query(User).filter(...)纯 SQL 操作你要用select([User]).where(...)。两套 API 各有各的语法各有各的踩坑姿势。新手经常搞混为什么query能直接拿到实体对象select返回的是一堆 Row2.0 把这个问题彻底解决了。你现在只有一个入口session.execute(select(User))返回的还是Result对象但通过.scalars()就能直接拿到实体通过.scalars().all()拿列表。Query这个类在 2.0 里被移到了遗留模块虽然from sqlalchemy.orm import Query还能用但它已经不再是文档推荐的主路径了。# 1.4 的老写法 users session.query(User).filter(User.age 18).all() # 2.0 的新写法 users session.scalars(select(User).where(User.age 18)).all()这个改动影响最大的不是语法是心智模型。你写 SQLAlchemy 2.0 的时候脑子里只需要记住一条线用select()构造语句用session.execute()执行语句。ORM 实体、普通列、聚合函数、子查询全部走同一条路。1.2 DeclarativeBase 终结了映射方式的混乱1.x 时代你可能有三种写法经典映射Tablemapper()、声明式映射declarative_base()、混合式__table__mapper。尤其老一辈代码里经典映射很常见因为那个年代Table对象是先用 SQL 定义好的mapper只是把类绑定上去。2.0 的DeclarativeBase把这个故事讲完了。现在定义一个基类子类就是映射模型。from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class User(Base): __tablename__ user_account id: Mapped[int] mapped_column(primary_keyTrue) name: Mapped[str] mapped_column(String(50))这里有个细节值得展开id: Mapped[int]这种类型标注不是给 IDE 看的摆设。SQLAlchemy 2.0 的整个映射系统是“类型驱动”的——它从Mapped[int]推断出这一列是INTEGER、非空、不带默认值。如果你写name: Mapped[str | None] mapped_column(String(50))它就自动知道这一列是NULL允许的。这就是 2.0 跟 1.4 最大的一点不同你用 Python 的类型语法描述数据库的结构。我个人的体会是这个改变带来的最大收益是重构安全。以前改一个字段类型要去翻数据库迁移脚本、模型定义、查询条件三处现在模型定义本身就是类型IDE 的 everywhere 搜索和类型检查能帮你在编译期之前发现大量错误。1.3 显式优于隐式Session 的过期与刷新策略变了2.0 对Session的改动很多人没注意到但实际影响非常大。老版本里session.commit()之前会自动执行一次flush把挂起的改动推给数据库。2.0 里这个行为被弱化了你主动session.flush()才会刷autoflush默认开启但在很多边界场景不再自动触发。更关键的是expire_on_commit的默认值没变但新引入了populate_existing之类的方法让你在处理“同一 Session 里对象可能过期”的场景时有了更明确的控制。说白了2.0 希望你清楚地知道你提交了什么、什么时候提交、提交后对象的状态是什么而不是让它自己去猜。这对异步项目尤其重要因为一旦涉及到await你根本不知道中间有没有别的协程碰了同一个对象。后面讲异步的时候再细说这里先记住一个基础结论2.0 的 Session 把你的手管得更紧了也更容易写出可预期、可调试的代码。2. ORM 映射实操Mapped 注解背后的细节画重点这一节是写代码时最容易被卡住的地方。2.1Mapped[...]与mapped_column()的分工很多新手会问既然有了Mapped[int]类型注解为什么还要mapped_column()因为类型注解描述的是“是什么”列约束描述的是“怎么存储”。Mapped[int]只是说“这属性是 int 类型、非空、映射到数据库 int 列”但你想设置primary_key、index、default、comment这些数据库层面的细节必须通过mapped_column()传进去。class User(Base): __tablename__ user id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) email: Mapped[str] mapped_column(String(255), uniqueTrue, indexTrue) age: Mapped[int | None] mapped_column(default0, comment年龄允许为空)Mapped[int | None]的写法要注意对于可空列类型注解里必须写| None否则 SQLAlchemy 会在建表时生成NOT NULL然后跟你的数据直接打架。这个坑我刚开始迁移的时候踩过好多次尤其是把老代码里nullableTrue转过来时漏写一个| None测试数据一组装就报IntegrityError。2.2 relationship 的注解写法关系映射是 ORM 里最值钱的部分。2.0 的relationship()也可以配合Mapped做类型注解让关系在 IDE 里点点就能跳转。class User(Base): __tablename__ user id: Mapped[int] mapped_column(primary_keyTrue) name: Mapped[str] addresses: Mapped[list[Address]] relationship(back_populatesuser) class Address(Base): __tablename__ address id: Mapped[int] mapped_column(primary_keyTrue) user_id: Mapped[int] mapped_column(ForeignKey(user.id)) email: Mapped[str] user: Mapped[User] relationship(back_populatesaddresses)两个细节值得注意。第一Mapped[list[Address]]这种写法是 Python 3.10 的list[...]泛型SQLAlchemy 能识别出这是一对多关系。第二back_populates两端都要写这是 2.0 推荐的做法。虽然backref还能用但它隐藏了很多内部逻辑调试的时候头大。2.3 type_annotation_map 处理自定义类型很多项目都有JSON字段、DateTime时区字段、Enum字段。1.x 时你可能习惯在Column里指定JSON()或者DateTime(timezoneTrue)可读性不错但重复劳动多。2.0 的type_annotation_map可以把“Python 类型”和“SQLAlchemy 类型”一次性映射好。from datetime import datetime from sqlalchemy import JSON, DateTime, String from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from typing import Any class Base(DeclarativeBase): type_annotation_map { dict[str, Any]: JSON, datetime: DateTime(timezoneTrue), str: String(255), } class Event(Base): __tablename__ event id: Mapped[int] mapped_column(primary_keyTrue) payload: Mapped[dict[str, Any]] created_at: Mapped[datetime]这样写完之后payload列自动就是 JSON 类型created_at自动就是带时区的DateTime。模型定义里几乎没有mapped_column(JSON)这种点了全是干净的Mapped[...]。代码读起来特别舒服像在读一份数据库 schema 的 Python 翻译。用法上有几个我实测的经验直接分享第一type_annotation_map是基类级别的所有继承它的模型都能吃到。如果个别模型想覆盖比如某个str字段想要TEXT而不是VARCHAR(255)直接在mapped_column()里传Text类型即可覆盖。第二dict[str, Any]这个映射在 PostgreSQL 场景会用到JSONB但 SQLAlchemy 默认的 JSON 类型在大部分数据库上都能工作除非你明确要JSONB否则不必过度设计。第二Enum 字段不建议放进type_annotation_map。原因很简单一个枚举类对应一个数据库枚举类型但你可能有多个不同的枚举类映射表里只能放一个“泛型入口”。我现在的做法是每个枚举字段在mapped_column()里显式写Enum(MyEnum)既清楚又不冲突。3. 查询与 Session从 query 到 select 的思维转换从 1.4 迁移到 2.0 的时候最难的不是模型定义而是把脑子里那套filter_by、order_by、.all()的肌肉记忆全部改掉。3.1select()的构造能力select()在 2.0 里是全能的。查全部列、查特定列、连表、子查询、聚合全部可以用统一的语法。# 查全部列返回 User 实体 stmt select(User).where(User.name.contains(张)) # 查特定列返回 Row 元组 stmt select(User.id, User.name).order_by(User.id.desc()) # 连表 stmt select(User, Address).join(Address).where(Address.email.endswith(example.com)) # 聚合 from sqlalchemy import func stmt select(func.count(User.id)).where(User.age 25)有几个细节新手容易踩where()里的走的是 Python 重载User.name 张三会被转成 SQL 的 张三而不是 Python 的布尔比较。这个知识 1.x 时代就有但 2.0 里where()因为没有Query.filter()那层包装语义更直接反而更好理解。join()默认用外键推断多个外键时须显式写join(Address, User.id Address.user_id)。select(User).where(User.age 18)返回的是Row对象里面第一个元素是User所以要用.scalars()解包。而select(User.id)返回的Row里第一个元素是int用.scalars().all()拿到[1, 2, 3]这样的列表这个用法写的多之后就会变成顺手。3.2 Session.execute() 的统一执行链路2.0 里Session的职责边界更清楚了它是事务和对象状态的容器不是查询构造器。所有语句都走session.execute(stmt)返回值是Result你根据语句类型决定是.scalars()、.mappings()还是.all()。实操上我总结了三句口诀查实体用session.scalars(select(User))查单行用session.scalar(select(func.max(User.id)))查多列用session.execute(select(User.id, User.name)).all()拿到的是Row元组列表# 单行标量 max_id session.scalar(select(func.max(User.id))) # 实体对象列表 users session.scalars(select(User).where(User.age 18)).all() # 多列元组列表 rows session.execute(select(User.id, User.name).where(User.age 18)).all() for user_id, name in rows: print(user_id, name)这个统一带来的一个额外好处是你可以把同一个select()语句直接丢给 Core 的连接去执行完全不依赖 ORM 的 Session。也就是下面这样from sqlalchemy import create_engine engine create_engine(sqlite:///test.db) with engine.connect() as conn: result conn.execute(select(User.id, User.name))在需要高性能批量查询、或者做报表类任务的时候这是极大的便利——你不用再维护两套 SQL 逻辑了。3.3 update 和 delete 走 Core 风格2.0 里Session的query(User).update(...)这种旧 API 没了更新删除统一走update()和delete()语句。from sqlalchemy import update, delete # 更新 stmt update(User).where(User.name old).values(namenew) result session.execute(stmt) # 删除 stmt delete(User).where(User.age 18) result session.execute(stmt) # 受影响行数 print(result.rowcount)注意update/delete默认是“同步更新数据库”的但如果你用 ORM 对象加载后再改字段那仍然是走flush的旧逻辑。这两者要注意区分别把批量语句和 ORM 对象状态搞混。4. 异步支持AsyncEngine 与 AsyncSession 全面上手这才是大家最关心的部分。SQLAlchemy 2.0 的异步支持终于不是“实验室”级别了官方文档里已经把它列为正式特性但实际操作起来还是有不少细节需要留意。4.1 异步引擎与会话工厂的搭建异步的核心是create_async_engine和async_sessionmaker。很多教程会说“只要把create_engine换成create_async_engine就行了”这话只说对了一半。你还得换数据库驱动同步的psycopg2没法在async环境里用需要换成asyncpgMySQL 要用aiomysql或者asyncmySQLite 可以用aiosqlite。from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker DATABASE_URL postgresqlasyncpg://user:passlocalhost:5432/mydb engine create_async_engine(DATABASE_URL, echoTrue, pool_size10, max_overflow20) AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse)这里的expire_on_commitFalse在异步环境里特别重要。如果你不设成 Falsecommit()之后对象会过期一旦你在await的过程中访问它的属性它就可能触发一次隐式的 SQL 查询。在异步里这种隐式 IO 会直接报错或者更隐蔽地造成锁等待。所以异步会话的标配是expire_on_commitFalse。4.2 AsyncSession 的用法完全版AsyncSession用法跟同步Session最大的区别就是凡是可能触发 IO 的操作都要await。async def get_user_by_id(user_id: int): async with AsyncSessionLocal() as session: stmt select(User).where(User.id user_id) result await session.execute(stmt) return result.scalars().first()这里要强调一个关键点AsyncSession不是让你把session.execute()前面加个await就完事了。它内部有严格的“只能被一个任务使用”的限制。什么意思就是同一个AsyncSession实例在一个时刻只能有一个协程在跑它的方法。官方文档管这叫“session 不能在多个任务之间共享”。举个例子如果你写async def concurrent_bad(session): await asyncio.gather( session.execute(select(User).where(User.id 1)), session.execute(select(User).where(User.id 2)), )这样在异步环境里几乎必然出问题因为两个任务同时在用一个 session 的底层连接。正确做法是每个任务独立创建 sessionasync def get_user(user_id: int): async with AsyncSessionLocal() as session: result await session.execute(select(User).where(User.id user_id)) return result.scalars().first() async def concurrent_good(): results await asyncio.gather( get_user(1), get_user(2), )这个原则跟“连接池”不是一回事是AsyncSession自身的事务状态机决定的。Session 里有事务、有 identity map你让两个协程同时改它的状态不出问题才怪。4.3 异步查询别偷懒使用隐式 IO2.0 的异步 ORM 有一条铁律不能有隐式 IO。同步时代你写user.addresses它会偷偷发一条 SQL 查关系数据这在异步环境是禁止的。如果你在异步代码里访问一个没加载的relationship属性你会看到类似MissingGreenlet的报错或者直接告诉你“cant load attribute, use await”。要正确地加载关系你有两个选择。选择一用selectinload显式预加载。from sqlalchemy.orm import selectinload stmt select(User).options(selectinload(User.addresses)).where(User.id 1) result await session.execute(stmt) user result.scalars().first() # 这时可以安全访问 user.addresses选择二用await session.refresh(instance)手动加载。这个一般用得少因为selectinload已经覆盖了绝大多数场景。我刚接触异步 ORM 的时候最大的一个错觉是“代码确实能跑通但报错总在奇怪的地方——比如序列化响应的时候或者在模板里渲染用户信息的时候”。后来才明白这些报错的根源都是隐式 IO。把selectinload加到位之后整个逻辑一下子就干净了。4.4 异步事务边界与连接池调优你写完async with AsyncSessionLocal() as session:之后事务的边界是进入上下文时开启事务正常退出时commit异常退出时rollback。这是一个很清爽的模型比手动begin/commit/rollback好维护得多。但注意async with自己不会帮你关闭底层连接它只是结束事务。连接池方面的调优要自己来。engine create_async_engine( DATABASE_URL, echoFalse, pool_size20, max_overflow10, pool_timeout30, pool_recycle1800, )pool_size建议根据asyncio.Semaphore控制的并发数来设置不要简单等于你的并发量。举个例子API 的并发是 100但数据库操作真正同时进行的一般只有 20~30连接池设太大反而浪费。还有一个很实用的经验pool_recycle一定要比数据库服务器的wait_timeout小不然空闲连接被服务端断开后客户端还傻傻地拿着用第一次查询就会报“connection already closed”。我一般都设成 1800 秒因为大部分 MySQL/PostgreSQL 默认wait_timeout是 8 小时。4.5 异步代码里跑同步阻塞操作的坑asyncpg是纯异步驱动全程无阻塞这很好。但你的项目里八成有别的同步库比如httpx的同步版本、boto3、或者某些自己不提供异步实现的工具库。在异步视图里直接用这些会导致事件循环被卡住。遇到这种情况别把数据访问和外部调用混在一个协程里要么用asyncio.to_thread把这些同步操作丢到线程池要么用独立的事件循环去跑。注意SQLAlchemy 的异步是用greenlet实现的它和asyncio.to_thread之间没有直接冲突但你把一个同步 ORM Session 丢进asyncio.to_thread是不合适的因为同步 Session 不是线程安全的。我的底线就是数据库操作走 async 原生链其他同步 IO 走线程池两者用数据组装不共享状态。5. 迁移避坑从 1.x 到 2.0 的常见问题排查最后这部分是给已经在迁移路上的开发者的我整理了最常遇到的几类问题每个都能让你卡上半天。5.1 Query 被移除后的替代写法速查1.4 旧写法2.0 新写法session.query(User).first()session.scalar(select(User))session.query(User).all()session.scalars(select(User)).all()session.query(User).filter_by(namea)select(User).where(User.name a)session.query(User).get(id)session.get(User, id)session.query(User).join(Address)select(User).join(Address)session.query(User).count()select(func.count()).select_from(User)session.get(User, id)这个新 API 比 1.x 的get更稳直接走主键查询而且在 Session 的 identity map 里如果有已经加载过的对象它直接返回不发 SQL。5.2 结果对象到底是 Row 还是实体这可能是新手最高频的问题。记住一个判断规则select(Entity)→ 结果为 Row用.scalars()取实体select(Entity.column1, Entity.column2)→ 结果为 Row直接.all()拿到元组列表select(func.count())→ 结果为 Row用.scalar()取单个值最容易翻车的地方是session.execute(select(User)).first()会得到一个RowRow[0]才是User。如果你直接对 Row 用 ORM 属性会报AttributeError。建议统一用.scalars()。5.3 autoflush 行为变了别依赖巧合1.4 里你在查询前session.add(obj)然后直接session.query(...)因为autoflush默认 True新对象会被自动刷进数据库。2.0 里这个行为大体保持但如果你在Session外面自己构造了select()再丢进来autoflush的触发时机会有细微差别。最稳妥的做法是需要更新数据库时显式await session.flush()或者await session.commit()。5.4 关系懒加载在异步里的“绿色陷阱”前面提过MissingGreenlet我再展开一下。greenlet是 SQLAlchemy 异步实现的基础当你在异步代码里访问一个懒加载属性SQLAlchemy 会试图用一个同步的 IO 去加载数据这在异步事件循环里是危险的。2.0 的检测机制很敏感它会直接抛错而不是悄悄阻塞。遇到这个报错排查顺序是这样的先看select()里有没有.options(selectinload(...))如果对象是通过asyncio.gather并发加载的确认每个 session 不能共享如果只是在代码里调试时访问关系可以先await session.refresh(user, [addresses])5.5 警惕异步环境下的 session 泄漏同步环境里你忘了session.close()最多是连接池被占住。异步环境里如果一个协程在异常退出前没把 session 关掉它所占用的连接可能被池子回收但事务状态还滞留。最直接的表现就是“concurrent tasks 卡死”或者“连接数飙升”。我的习惯是async with上下文管理器能覆盖 99% 的场景但如果你在类内部长期持有 session比如一个长连接的服务必须实现__aenter__/__aexit__来保证异常时关闭。别裸用一个 session 到处传。一点最后的经验SQLAlchemy 2.0 的迁移工具链上很成熟但心智迁移是最大的成本。我的建议是第一周别急着改代码先把一张速查表放桌面比如把query.filter对应到select.where、把one_or_none对应到session.scalar。哪怕你写的代码是 1.4 风格只要在 2.0 环境里运行也会有一堆废弃警告逐条清掉它们就是最好的练习。项目里如果还有大量session.query的调用可以先用sqlalchemy.orm.Query的遗留兼容模式跑测试稳定之后再逐步替换。千万不要一把梭全改测试覆盖不够的项目会改出满地鸡毛。另外一个我一直保留的习惯是每次写完一段 ORM 操作都打开数据库的 query log 看生成的 SQL。SQLAlchemy 2.0 生成的 SQL 比 1.4 干净很多但也更容易因为selectinload而多出几条查询。看着实际下发的 SQL 来调代码比对着文档猜要快得多。特别是异步场景多一条隐式查询就可能让并发性能掉一个量级。SQLAlchemy 2.0 不是那种“升完级啥也没变”的版本它是值得你把整个数据访问层重写一遍的版本。不是因为旧的不行而是新的这套类型驱动、统一查询、显式异步的模型确实能让代码在长时间维护里少操很多心。