Async SQLAlchemy在pytest fixture中报错:greenlet_spawn未调用
解决SQLAlchemy异步Pytest Fixture中的MissingGreenlet错误
问题原因
你遇到的sqlalchemy.exc.MissingGreenlet错误,本质是当异步SQLAlchemy Session的事务上下文结束后,尝试访问ORM对象的属性触发了延迟加载,但此时没有活跃的greenlet上下文支持数据库操作。
在你的test_users fixture中,async with db_session.begin()会创建一个临时事务,当fixture执行完毕返回用户列表时,这个事务已经结束;而another_fixture中访问test_users[0].id时,ORM对象仍会尝试关联Session进行验证,此时Session已不在活跃的异步上下文里,就会抛出该错误。
解决方法
方法1:将ORM对象从Session中分离(脱管)
在test_users fixture中,查询完成后将用户对象从Session中分离,使其不再依赖Session上下文,访问属性时不会触发数据库操作:
from typing import List from your_module import User import pytest_asyncio @pytest_asyncio.fixture(scope="function") async def test_users(db_session): async with db_session.begin(): users = await db_session.scalars(select(User)) users_list = users.all() # 分离所有用户对象,解除与Session的关联 for user in users_list: await db_session.expunge(user) return users_list
方法2:提前加载所需属性,避免延迟加载
通过load_only明确指定要加载的字段,确保属性在查询时已被加载,后续访问无需触发数据库查询:
from typing import List from your_module import User from sqlalchemy.orm import load_only import pytest_asyncio @pytest_asyncio.fixture(scope="function") async def test_users(db_session): async with db_session.begin(): cursor = await db_session.execute( select(User).options(load_only(User.id)) # 明确加载id字段 ) return cursor.scalars().fetchall()
额外调整方案(可选)
如果需要保持事务覆盖测试全程,可以将test_users的返回改为yield,让事务保持到fixture被使用完毕(注意可能影响测试隔离性):
@pytest_asyncio.fixture(scope="function") async def test_users(db_session): async with db_session.begin(): users = await db_session.scalars(select(User)) users_list = users.all() yield users_list # 用yield替代return,事务会保持到fixture生命周期结束
内容的提问来源于stack exchange,提问作者Suoslex
相关产品推荐
相关产品推荐

