中文
devexamples

Motor Async MongoDB Aggregation Pipeline

Run aggregation pipelines with the async Motor driver, streaming large result sets via async iteration and cursor batching.

PythonData Fetching

Code

snippet
import asyncio

from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient('mongodb://localhost:27017')
orders = client['shop']['orders']


PIPELINE = [
    {'$match': {'status': 'paid'}},
    {'$unwind': '$items'},
    {'$group': {
        '_id': '$items.sku',
        'units': {'$sum': '$items.qty'},
        'revenue': {'$sum': {'$multiply': ['$items.qty', '$items.price']}},
    }},
    {'$sort': {'revenue': -1}},
    {'$limit': 20},
]


async def top_products() -> list[dict]:
    cursor = orders.aggregate(
        PIPELINE,
        batchSize=200,
        allowDiskUse=True,
    )

    results: list[dict] = []
    async for doc in cursor:
        results.append({
            'sku': doc['_id'],
            'units': doc['units'],
            'revenue': round(doc['revenue'], 2),
        })
    return results


async def seed() -> None:
    await orders.delete_many({})
    await orders.insert_many([
        {'status': 'paid', 'items': [
            {'sku': 'A-1', 'qty': 2, 'price': 10.0},
            {'sku': 'B-2', 'qty': 1, 'price': 30.0},
        ]},
        {'status': 'paid', 'items': [
            {'sku': 'A-1', 'qty': 5, 'price': 10.0},
        ]},
        {'status': 'refunded', 'items': [
            {'sku': 'C-3', 'qty': 1, 'price': 99.0},
        ]},
    ])


async def main() -> None:
    await seed()
    for row in await top_products():
        print(row)
    client.close()


asyncio.run(main())

Related Examples