mongodb

MongoDB — Async Patterns with Motor

Async MongoDB via Motor, aggregation pipelines, and index design.

When to Activate

• Writing async MongoDB queries with Motor
• Designing aggregation pipelines ($match, $group, $lookup, $unwind)
• Creating indexes (compound, text, TTL, sparse, partial)
• Running multi-document transactions
• Watching for real-time changes with change streams
• Working with adk.state (Agentex per-task state backed by MongoDB)
• Designing document schemas for flexible or hierarchical data

---

Connection

from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase

client = AsyncIOMotorClient("mongodb://localhost:27017")
db: AsyncIOMotorDatabase = client["mydb"]

# With auth + replica set (production)
client = AsyncIOMotorClient(
    "mongodb://user:pass@host1:27017,host2:27017/mydb?replicaSet=rs0&authSource=admin"
)

# Close on shutdown
client.close()

Collections are accessed as attributes — no schema declaration needed:

users = db["users"]              # or db.users
orders = db.orders

---

CRUD

from datetime import datetime, timezone
from bson import ObjectId

# Insert one
result = await db.users.insert_one({
    "email": "[email protected]",
    "name": "Alice",
    "role": "user",
    "created_at": datetime.now(timezone.utc),
})
inserted_id = result.inserted_id   # ObjectId

# Insert many
result = await db.users.insert_many([
    {"email": "[email protected]", "name": "Bob"},
    {"email": "[email protected]", "name": "Carol"},
])

# Find one
user = await db.users.find_one({"email": "[email protected]"})
user = await db.users.find_one({"_id": ObjectId("64a...")})

# Find many — returns an async cursor
cursor = db.users.find({"role": "admin"}).sort("created_at", -1).skip(0).limit(20)
users = await cursor.to_list(length=None)   # length=None = all results

# Count
count = await db.users.count_documents({"role": "admin"})
estimated = await db.users.estimated_document_count()   # fast, uses metadata

# Update one
result = await db.users.update_one(
    {"_id": ObjectId("64a...")},
    {"$set": {"role": "admin", "updated_at": datetime.now(timezone.utc)}},
)
matched = result.matched_count
modified = result.modified_count

# Update many
await db.users.update_many(
    {"role": "user", "created_at": {"$lt": cutoff_date}},
    {"$set": {"tier": "legacy"}},
)

# Upsert
await db.users.update_one(
    {"email": "[email protected]"},
    {"$setOnInsert": {"created_at": datetime.now(timezone.utc)},
     "$set": {"name": "Dave", "role": "user"}},
    upsert=True,
)

# Delete
await db.users.delete_one({"_id": ObjectId("64a...")})
await db.users.delete_many({"status": "inactive", "created_at": {"$lt": cutoff}})

# Find one and update (atomic — returns updated doc)
updated = await db.users.find_one_and_update(
    {"_id": ObjectId("64a...")},
    {"$inc": {"login_count": 1}},
    return_document=True,    # return doc after update
)