Motor 异步聚合管道

示例说明

聚合管道把统计下推到数据库侧执行：match 先过滤缩小数据面，unwind 把订单行展开成单品记录，group 按 SKU 汇总销量与营收，最后排序取头部。Motor 是 PyMongo 的异步孪生——同样的管道语法，但游标以异步迭代消费，分批取数让大结果集不会一次性占满内存。

技术要点

• 管道各阶段服务端顺序执行
• async for 流式消费游标
• allowDiskUse 允许大分组溢出到磁盘

代码解读

PIPELINE 五个阶段构成畅销品统计；aggregate 的 batchSize 控制单批拉取量，allowDiskUse 给超内存分组留出口。seed 插入三类订单：两笔已付含三个商品行、一笔退款用于验证 match 过滤。A-1 商品合计七件营收七十元应排首位。

使用建议

match 尽量前置并命中索引。 unwind 后数据量按行数膨胀，评估管道成本再上线。游标遍历中处理要快，慢消费会拖住服务端游标。

代码示例

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())