如何在Python Flask中使用SQLAlchemy 2.0实现异步查询？-Python教程-PHP中文网

不能直接套用老写法；async_sessionmaker返回协程对象，必须await才能获取asyncsession实例，所有操作如execute、commit均需await，且须搭配create_async_engine和异步驱动（如postgresql+asyncpg://），禁用同步引擎与sessionmaker。

如何在python flask中使用sqlalchemy 2.0实现异步查询？

SQLAlchemy 2.0 的 async_sessionmaker 能直接用吗？

不能直接套用老写法。Flask 默认是同步框架，async_sessionmaker 返回的是协程对象，如果在普通视图函数里 await 它，会报 RuntimeWarning: coroutine 'AsyncSession.<strong>anext</strong>' was never awaited 或直接抛出 RuntimeError: no running event loop。根本原因是 Flask 2.2+ 虽支持 async def 视图，但底层 WSGI 服务器（如 Werkzeug 开发服务器）不运行 asyncio event loop —— 除非你显式启用异步支持。

必须用 app.run(..., use_reloader=False) 启动（reloader 会破坏 event loop）
推荐换用 ASGI 服务器，比如 uvicorn：运行 uvicorn app:app，而非 flask run
Flask 视图函数必须声明为 async def，否则 await 语法非法

如何配置支持 await 的 AsyncEngine 和 AsyncSession？

核心是绕过 create_engine，改用 create_async_engine，并搭配 async_sessionmaker。同步的 engine 和 sessionmaker 对 async 查询完全无效。

URL 必须带异步驱动前缀，例如 PostgreSQL 用 postgresql+asyncpg://，SQLite 不支持异步，别试
echo=True 在 async 模式下仍可用，但日志可能交错，建议上线关掉
async_sessionmaker 的 expire_on_commit=False 是必须的，因为 async session 不支持 commit 后自动 expire（会报 AttributeError: 'AsyncSession' object has no attribute '_remove_newly_added'）

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
<p>engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
echo=True,
pool_pre_ping=True,
)
AsyncSessionLocal = async_sessionmaker(
bind=engine,
expire_on_commit=False,
)</p>

在 Flask 路由里怎么安全 await 查询？

不能在普通 def 视图里 await；也不能把 AsyncSessionLocal() 当成同步对象调用。常见错误包括：

忘记加 async 前缀，导致 SyntaxError: 'await' outside async function
手动调用 session.close() —— 应该用 async with session: 确保正确释放连接
直接对模型实例调 .to_dict() 类方法，而该方法内部用了同步的 session.dirty 等属性（会触发 AttributeError）

正确的写法是：

python 查询技能

查询客流数据，输出JSON格式，可直接导入Bitable等可视化工具

下载

@app.route("/users")
async def get_users():
    async with AsyncSessionLocal() as session:
        stmt = select(User).where(User.active == True)
        result = await session.execute(stmt)
        users = result.scalars().all()
        return {"users": [u.to_dict() for u in users]}

注意：scalars().all() 是 awaitable，必须 await；result.all() 也行，但返回的是元组，不如 scalars() 直观。

为什么 await session.get() 有时返回 None，即使 ID 存在？

这不是 bug，是 async session 的缓存行为和生命周期导致的典型问题。session.get() 默认走 identity map 缓存，但 async session 的缓存只在当前 async with 块内有效；跨请求不会共享，这点和 sync session 一样。真正容易踩的坑是：

在同一个 session 中混用 get() 和 execute(select(...))，且未 await 后者，导致 session 状态错乱
使用 session.get(User, 123) 时没加 await，结果得到一个 coroutine 对象，被隐式转为 None（尤其在 JSON 序列化时静默失败）
PostgreSQL 的 asyncpg 驱动对某些时间类型（如 timetz）解析异常，可能让整行数据加载失败，表现为“查不到”