Skip to content

Use Case 19: Bulk Operations

Scenario: A background job needs to update or delete hundreds of documents in one call — doing it one-by-one would be prohibitively slow.

📦 What's New?

Component Description
update_many / delete_many Motor's multi-document write operations.
Repository wrapper Keeps bulk logic behind a domain method so callers pass IDs and business parameters, not raw dicts.
BulkWriteError Raised when ordered=False and some operations fail — catch it for partial-success handling.

🚀 Example

Python
from datetime import datetime

from bson import ObjectId, BulkWriteError
from fastapi import FastAPI, HTTPException
from mongo_ops import BaseDocument, BaseRepository


# ---------------------------
# Model
# ---------------------------
class Task(BaseDocument):
    status: str = "pending"
    assignee_id: str | None = None
    priority: int = 0


# ---------------------------
# Repository
# ---------------------------
class TaskRepository(BaseRepository[Task]):
    def __init__(self):
        super().__init__("tasks", Task)

    async def bulk_update_status(self, ids: list[str], status: str) -> int:
        """Set status on many tasks at once. Returns the count of modified documents."""
        object_ids = [ObjectId(i) for i in ids]
        result = await self.collection.update_many(
            {"_id": {"$in": object_ids}},
            {"$set": {"status": status, "updated_at": datetime.utcnow()}},
        )
        return result.modified_count

    async def bulk_reassign(self, old_assignee: str, new_assignee: str) -> int:
        """Move all pending tasks from one assignee to another."""
        result = await self.collection.update_many(
            {"assignee_id": old_assignee, "status": "pending"},
            {"$set": {"assignee_id": new_assignee, "updated_at": datetime.utcnow()}},
        )
        return result.modified_count

    async def bulk_archive(self, ids: list[str]) -> int:
        """Delete many tasks at once. Returns the count of deleted documents."""
        object_ids = [ObjectId(i) for i in ids]
        result = await self.collection.delete_many({"_id": {"$in": object_ids}})
        return result.deleted_count

    async def bulk_upsert_users(self, records: list[dict]) -> dict:
        """Insert-or-update many user records. Returns inserted/modified counts."""
        from pymongo import UpdateOne

        ops = [
            UpdateOne(
                {"email": rec["email"]},
                {"$set": rec},
                upsert=True,
            )
            for rec in records
        ]
        try:
            result = await self.collection.bulk_write(ops, ordered=False)
            return {
                "inserted": result.upserted_count,
                "modified": result.modified_count,
            }
        except BulkWriteError as exc:
            return {
                "inserted": exc.details.get("upsertedCount", 0),
                "modified": exc.details.get("modifiedCount", 0),
                "errors": exc.details.get("writeErrors", []),
            }


app = FastAPI()
task_repo = TaskRepository()


@app.post("/tasks/bulk-status")
async def bulk_status(ids: list[str], status: str):
    count = await task_repo.bulk_update_status(ids, status)
    return {"modified": count}


@app.post("/tasks/bulk-reassign")
async def reassign(old_assignee: str, new_assignee: str):
    count = await task_repo.bulk_reassign(old_assignee, new_assignee)
    return {"modified": count}


@app.post("/tasks/bulk-archive")
async def archive(ids: list[str]):
    count = await task_repo.bulk_archive(ids)
    return {"deleted": count}

Note: lifespan wiring omitted — copy from use case 01.

💡 Tips

  • Bulk methods bypass the caching and population layers. If you need cache consistency after a bulk write, call invalidate_cache(id) for each affected document, or reinitialize the cache (see use case 11).
  • Use ordered=False for upserts so one bad record doesn't block the rest. Catch BulkWriteError and surface the per-operation errors.
  • For very large batches (10k+), chunk the operation into pages of 1 000 to avoid exceeding MongoDB's 16 MB bulkWrite payload limit.
  • update_many / delete_many require a {"_id": {"$in": [...]}} filter to scope the operation — never pass an unscoped filter.