直接用 create_async_engine 配置 asyncpg + SQLAlchemy 2.0,URL 必须为 postgresql+asyncpg://,poolclass 设为 NullPool,且必须显式 await engine.dispose()。

asyncpg + SQLAlchemy 2.0 的 async engine 怎么配
直接用 create_async_engine,别碰 create_engine —— 后者返回同步 engine,哪怕 URL 带 postgresql+asyncpg:// 也会在第一次 await conn.execute(...) 时抛 RuntimeError: This event loop is already running 或静默阻塞。
关键点:
-
url必须以postgresql+asyncpg://开头,不能用psycopg或pg8000的 async 变体(它们不被 SQLAlchemy 2.0 官方 async flow 支持) -
poolclass要设为NullPool(默认是AsyncAdaptedQueuePool,但实际连接复用由 asyncpg 内部管理,SQLAlchemy 的 pool 层只是透传) - 必须显式
await engine.dispose()关闭,否则 test 环境常因连接未释放导致too many clients
示例片段:
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
echo=True,
poolclass=NullPool, # 避免双重 pooling
)
用 GINO 还是 TortoiseORM?选型看这三点
GINO 已停止维护(最后 release 是 2021),TortoiseORM 活跃但默认用 aiosqlite 或 asyncpg 时行为不一致:比如 .all() 返回的是 list 还是 async generator,取决于是否加 await;而它的 select_related 在 PostgreSQL 下会生成 LEFT JOIN,但没做字段去重,容易引发 KeyError: 'id'。
实操建议:
- 新项目优先选
SQLModel+SQLAlchemy 2.0 async组合——类型提示友好,select()和session.exec()分离清晰,且@asynccontextmanager封装 session 很自然 - 若必须用 ORM 层自动建表,Tortoise 的
Tortoise.init()要等 event loop 启动后调用,不能放在模块顶层,否则报RuntimeError: There is no current event loop in thread - GINO 的
db.bind已被弃用,强行用会触发DeprecationWarning: bind is deprecated并在 1.1+ 版本失效
连接池满、timeout、transaction stuck 怎么查
根本原因不是并发高,而是异步任务没正确 await 或忘了 close —— 比如同一个 AsyncSession 被多个协程共享,或 session.begin() 后没配 await session.commit() / await session.rollback()。
快速定位手段:
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac,无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启:辅助功能 + 屏幕录制权限
- 查数据库当前连接:
SELECT * FROM pg_stat_activity WHERE state = 'active' OR state = 'idle in transaction';,重点关注backend_start和state_change时间差 - 在 engine 初始化时加
pool_pre_ping=True,避免 stale connection 导致的asyncpg.exceptions.InterfaceError: cannot perform operation: another operation is in progress - 用
asyncpg.Pool原生接口时,max_size默认是 10,但min_size是 0 —— 若短时 burst 请求多,会频繁创建/销毁连接,建议设min_size=5
FastAPI 里怎么安全注入 async DB session
别用依赖项返回 AsyncSession 实例——它不是线程安全,更不是协程安全;每次请求必须新建 session,用完即丢。常见错误是把 session 存在类属性或全局变量里,导致事务交叉污染。
正确姿势:
- 依赖函数里用
async_sessionmaker创建新 session,并用try/finally保证await session.close() - 不要在中间件里
await session.begin()—— FastAPI 的依赖解析顺序不保证中间件早于路由,容易漏 commit - 如果用
sqlmodel.Session,注意它包装的是AsyncSession,所以session.exec(select(User)).all()是同步调用,必须写成(await session.exec(select(User))).all()
最简依赖示例:
async def get_session():
async with AsyncSession(engine) as session:
yield session
这个 yield 是关键:FastAPI 会在响应返回后自动执行 session.close()。
异步 session 的生命周期管理比同步难在「看不见的等待」——一个没 await 的 session.execute 不报错,但会卡住整个连接,而且不会释放。调试时多打日志,少信“应该没问题”。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










