AttributeError: 'async_generator'对象无对应属性的原因分析

错误原因分析 #

1. AttributeError: 'async_generator' object has no attribute 'begin' #

这个错误说明你在异步生成器对象上直接调用了.begin()方法，但异步生成器本身并没有这个方法——你需要先获取它产出的AsyncSession实例，再进行事务操作。

通常是测试中错误地将数据库会话生成器（比如get_db()这类用async def + yield定义的函数）当成了实际的会话对象来使用。

2. AttributeError: 'async_generator' object has no attribute 'add' #

同理，这个错误是你直接在异步生成器上调用了.add()方法，而非生成器产出的AsyncSession实例。本质问题和第一个错误一致：没有正确获取异步会话对象。

修复步骤 #

1. 确认异步会话生成器的正确定义 #

确保你的数据库会话生成器是通过异步上下文管理器返回AsyncSession实例：

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# 初始化异步引擎和会话工厂
engine = create_async_engine("sqlite+aiosqlite:///./test.db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# 正确的异步会话生成器
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

2. 测试用例中正确获取会话 #

在测试代码中，必须通过异步上下文管理器或async for循环获取生成器产出的会话实例，再执行数据库操作：

# 示例测试用例
async def test_create_item():
    # 方式1：用async with获取会话
    async with get_db() as db:
        async with db.begin():
            new_item = Item(name="Test Item")
            db.add(new_item)
            await db.commit()
            await db.refresh(new_item)
            assert new_item.id is not None

    # 方式2：用async for循环获取会话
    async for db in get_db():
        async with db.begin():
            # 执行操作...
            pass

3. 检查测试配置的异步兼容性 #

• 确保使用create_async_engine创建异步引擎，而非同步的create_engine
• 会话工厂必须指定class_=AsyncSession，确保生成的是异步会话对象

关键注意点 #

异步SQLAlchemy的会话操作全程需要异步上下文（async with）和await关键字，不能像同步代码那样直接调用会话方法。如果测试中跳过了获取实际会话的步骤，直接操作生成器对象，就会触发这类AttributeError。

内容的提问来源于stack exchange，提问作者Kenneth Mamahit