Aggregation and bulk writes¶

Aggregation pipelines and bulk requests are native MongoDB/PyMongo structures. Mongoz selects the document's collection, propagates an optional session, closes aggregation cursors, and preserves native results and errors. It does not sanitize pipeline operators.

Aggregation¶

pipeline = [
    {"$match": {"status": "paid"}},
    {"$group": {"_id": "$customer_id", "total": {"$sum": "$amount"}}},
    {"$sort": {"total": -1}},
]

totals = await Invoice.aggregate(pipeline)

aggregate() materializes the complete result as a list of mappings. For unbounded pipelines or fine-grained cursor options, use Invoice.get_collection().aggregate(...) and own cursor cleanup.

Bulk writes¶

from pymongo import DeleteOne, UpdateOne

result = await Invoice.bulk_write(
    [
        UpdateOne({"status": "draft"}, {"$set": {"status": "expired"}}),
        DeleteOne({"status": "cancelled", "retention_complete": True}),
    ],
    ordered=True,
)

print(result.modified_count, result.deleted_count)

bulk_write() accepts PyMongo collection-level write models and returns BulkWriteResult. BulkWriteError, duplicate-key failures, and write-concern failures propagate unchanged.

Both methods accept session=. The sessions and transactions tutorial shows the required propagation pattern.