使用TaskGroup结合Motor Aggregate时遭遇StopAsyncIteration错误

云原生阿岚

2026-6-23

使用TaskGroups迭代Motor聚合结果时出现StopAsyncIteration错误

在使用asyncio.TaskGroups并行迭代Motor的聚合查询结果时，会抛出StopAsyncIteration错误，但串行执行或先串行后并行的场景下无此问题。

复现代码

import asyncio

from motor.motor_asyncio import AsyncIOMotorClient


def create_aggregate():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    collection = client["test"]["test"]

    return collection.aggregate([])

async def get_next(agg):
    item = await agg.next()
    print(item)

async def run_1_success():
    agg = create_aggregate()
    
    await get_next(agg)

async def run_2_success():
    agg = create_aggregate()
    await get_next(agg)

    async with asyncio.TaskGroup() as tg:
        tg.create_task(get_next(agg))
        tg.create_task(get_next(agg))

async def run_3_error():
    agg = create_aggregate()

    async with asyncio.TaskGroup() as tg:
        tg.create_task(get_next(agg))
        tg.create_task(get_next(agg))