SQLAlchemy 2.0 异步需用 create_async_engine + 异步驱动 URL（如 postgresql+asyncpg://），配 AsyncEngine 和 AsyncSession；所有数据库操作必须 await，结果方法（如 all()、scalars()）也须 await，不可混用同步方式。
SQLAlchemy 2.0 的 async_engine 必须用 create_async_engine
直接复用 create_engine 会报错 RuntimeError: This event loop is already running 或根本无法 await —— 因为它返回的是同步引擎。必须显式调用 create_async_engine，且 URL 协议头要换成异步驱动，比如 postgresql+asyncpg://、mysql+aiomysql://，不能用 sqlite:///（无官方异步 SQLite 支持）。
常见错误是照搬旧代码改 import 就完事，结果卡在 session.execute() 不返回。检查你的 engine 实例类型：isinstance(engine, AsyncEngine) 应为 True。
-
create_async_engine返回的AsyncEngine不支持.connect()同步方法，只能用.connect()（返回AsyncConnection）或直接进AsyncSession - URL 中若漏写
+asyncpg（只写postgresql://），会静默回退到同步驱动，但后续 await 会抛AttributeError: 'Connection' object has no attribute '__await__' - 连接池参数如
poolclass=NullPool在异步场景下仍是推荐的，避免多线程/事件循环混用问题
AsyncSession 才能执行 awaitable 查询
session.execute()、session.scalars()、session.get() 这些方法在 AsyncSession 下才返回可 await 对象；用普通 Session 会直接报 TypeError: object AsyncSession can't be used in 'await' expression（实际是方法没被 await 化）。
别试图把 AsyncSession 当成普通 session 传给老代码——它的所有数据库操作都必须加 await，且生命周期需配合 async context manager 或手动 await session.close()。
- 创建方式：用
async_sessionmaker绑定AsyncEngine，而非sessionmaker -
await session.execute(select(User).where(User.id == 1))是合法的；但session.execute(...).scalars().all()不行——得写成(await session.execute(...)).scalars().all()或更推荐await session.scalars(...).all() -
await session.commit()是必须的，rollback()同理；没有自动 commit，也不会因函数退出隐式提交
异步查询结果不能直接 .all() 或 list()，得 await .all()
很多人卡在这一步：写了 result = await session.execute(stmt)，然后想 list(result) 或 result.all()，结果报 TypeError: 'Result' object is not iterable —— 因为 Result 对象本身不可迭代，它的 .all()、.first()、.scalars().all() 都是协程方法，必须 await。
尤其注意 scalars() 返回的是 ScalarResult，它的 .all() 也得 await，不是同步取值。
- 正确写法：
users = await session.scalars(select(User)).all()，不是session.scalars(...).all()（少 await） - 如果只取一个对象，
user = await session.get(User, 1)比await session.execute(...).scalar_one_or_none()更简洁安全 - fetch 大量数据时，避免
await result.all()一次性加载进内存；可用await result.stream_scalars()+ 异步 for 循环流式处理
Python 3.11 的 asyncio.run() 和嵌套事件循环陷阱
在脚本顶层直接写 async def main(): ...; asyncio.run(main()) 没问题，但一旦你把它塞进 Jupyter、某些测试框架或已有事件循环的环境（如 FastAPI 的依赖函数里），asyncio.run() 会报 RuntimeError: asyncio.run() cannot be called from a running event loop。
这时候得用 asyncio.create_task() 或确保外层已提供运行环境。SQLAlchemy 自身不启动事件循环，它只依赖当前活跃的 loop。
- 在 FastAPI 路由中，直接
await即可，不用asyncio.run() - Jupyter 中建议用
await表达式（IPython 8.0+ 支持），而不是封装成asyncio.run() - 测试时若用 pytest-asyncio，确保 fixture 返回的是协程，并用
@pytest.mark.asyncio标记测试函数
真正容易被忽略的是：异步不是加个 async/await 就完事。从驱动安装（pip install asyncpg）、URL 协议、引擎类型、session 类型，到结果处理方式，每个环节断链都会让查询“看起来没反应”或抛出看似不相关的异常。最稳妥的验证路径是先跑通一个 await engine.connect()，再加 await conn.execute(text("SELECT 1"))，最后才引入 ORM 层。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
