如何在Python3.10中测试SQLAlchemy异步会话-Python教程-PHP中文网

sqlalchemy 3.0+ 异步会话必须用 create_async_engine 初始化，混用同步 create_engine 会触发 attributeerror 或 notimplementederror；需配异步驱动url、async_sessionmaker 工厂及 await session.commit() 等规范操作。

如何在python3.10中测试sqlalchemy异步会话

SQLAlchemy 3.0+ 异步会话必须用 `create_async_engine` 初始化

直接复用同步 create_engine 会导致 AttributeError: 'Engine' object has no attribute 'connect' 或运行时抛出 NotImplementedError: This engine is not configured for asynchronous execution。异步会话依赖底层引擎的协程能力，而同步引擎压根没实现 connect() 的异步版本。

实操建议：

python全能编程助手

SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的双功能合体技能，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途：；核心要点📝 题解模式：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式：输入 Python 代码，自动添加详细中。它将相关步骤、

下载

导入必须用 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession，不能混用 sqlalchemy.create_engine
数据库 URL 必须带异步驱动：PostgreSQL 用 postgresql+asyncpg://，MySQL 用 mysql+aiomysql://（需额外装 aiomysql）
禁用 echo=True 以外的日志选项（如 echo_pool），某些异步驱动不支持

测试中正确构造 `AsyncSession` 实例

不能直接 session = AsyncSession(...) —— 这会跳过引擎绑定和事务上下文管理，导致 session.execute() 报 InvalidRequestError: This Session does not support the use of 'execute()'。

实操建议：

用 async_sessionmaker 工厂封装：

engine = create_async_engine("sqlite+aiosqlite:///test.db", echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

测试函数内用 async with AsyncSessionLocal() as session: 获取实例，确保自动清理连接
手动 commit/rollback 必须显式调用 await session.commit() 或 await session.rollback()，不会自动触发

pytest 中运行 async 测试函数要配 `pytest-asyncio`

默认 pytest 不识别 async def test_xxx()，会跳过或报 RuntimeWarning: coroutine 'test_xxx' was never awaited。

实操建议：

安装 pip install pytest-asyncio
在测试文件顶部加 @pytest.mark.asyncio 装饰器，或全局配置 pytest.ini 启用自动发现：
```
[tool:pytest]
asyncio_mode = auto
```
避免在 setup_method 等同步钩子里 await 异步操作；全部逻辑移到带 @pytest.mark.asyncio 的测试函数内

Mock 异步查询结果时别碰 `AsyncSession.execute` 的返回类型

直接 patch AsyncSession.execute 返回普通 Result 会导致 TypeError: object Result can't be used in 'await' expression —— 因为真实调用返回的是 AsyncResult，它实现了 __await__。

实操建议：

mock 时返回一个可 await 的对象：

from unittest.mock import AsyncMock
mock_result = AsyncMock()
mock_result.scalars.return_value.all.return_value = [User(id=1)]
session.execute = AsyncMock(return_value=mock_result)