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": "alice@example.com",
    "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": "bob@example.com", "name": "Bob"},
    {"email": "carol@example.com", "name": "Carol"},
])

# Find one
user = await db.users.find_one({"email": "alice@example.com"})
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": "dave@example.com"},
    {"$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
)

Query Operators

# Comparison
{"age": {"$gt": 18, "$lte": 65}}
{"status": {"$in": ["active", "pending"]}}
{"status": {"$nin": ["banned", "deleted"]}}
{"score": {"$ne": 0}}

# Logical
{"$and": [{"role": "admin"}, {"active": True}]}
{"$or":  [{"email": {"$regex": "@company.com"}}, {"role": "admin"}]}
{"$not": {"status": "banned"}}

# Array operators
{"tags": {"$all": ["python", "async"]}}        # array contains all
{"tags": {"$elemMatch": {"$gt": 10, "$lt": 20}}}  # element matching condition
{"tags.2": "python"}                           # index access

# Element operators
{"phone": {"$exists": True}}
{"age":   {"$type": "int"}}

# Regex
{"email": {"$regex": "^admin", "$options": "i"}}

# Nested document
{"address.city": "New York"}
{"address.zip": {"$in": ["10001", "10002"]}}

Update Operators

# $set — update or add fields
{"$set": {"name": "Alice", "role": "admin"}}

# $unset — remove fields
{"$unset": {"temp_token": "", "legacy_field": ""}}

# $inc — atomic increment
{"$inc": {"login_count": 1, "score": -5}}

# $push — append to array
{"$push": {"tags": "python"}}
{"$push": {"events": {"$each": ["a", "b"], "$slice": -100}}}  # keep last 100

# $addToSet — append only if not present (unique set)
{"$addToSet": {"permissions": "write"}}

# $pull — remove from array
{"$pull": {"tags": "deprecated"}}
{"$pull": {"events": {"type": "click"}}}   # remove matching sub-docs

# $setOnInsert — only set on upsert insert (not on update)
{"$setOnInsert": {"created_at": datetime.now(timezone.utc)}}

Aggregation Pipeline

# Basic aggregation — group orders by status with total revenue
pipeline = [
    {"$match": {"created_at": {"$gte": start_date}}},
    {"$group": {
        "_id": "$status",
        "count": {"$sum": 1},
        "total_revenue": {"$sum": "$total"},
        "avg_order": {"$avg": "$total"},
    }},
    {"$sort": {"total_revenue": -1}},
]
results = await db.orders.aggregate(pipeline).to_list(None)

# $lookup — JOIN equivalent
pipeline = [
    {"$match": {"role": "admin"}},
    {"$lookup": {
        "from": "orders",           # collection to join
        "localField": "_id",        # field from users
        "foreignField": "user_id",  # field from orders
        "as": "orders",             # output array field
    }},
    {"$addFields": {"order_count": {"$size": "$orders"}}},
    {"$project": {"name": 1, "email": 1, "order_count": 1, "_id": 0}},
]

# $unwind — flatten array field into separate documents
pipeline = [
    {"$unwind": "$items"},          # one doc per item
    {"$group": {
        "_id": "$items.product_id",
        "total_sold": {"$sum": "$items.quantity"},
    }},
]

# $facet — multiple aggregations in one query
pipeline = [
    {"$match": {"status": "active"}},
    {"$facet": {
        "by_role": [
            {"$group": {"_id": "$role", "count": {"$sum": 1}}},
        ],
        "total": [
            {"$count": "count"},
        ],
        "recent": [
            {"$sort": {"created_at": -1}},
            {"$limit": 5},
            {"$project": {"name": 1, "email": 1}},
        ],
    }},
]

# $bucket — range bucketing
pipeline = [
    {"$bucket": {
        "groupBy": "$total",
        "boundaries": [0, 50, 100, 500, 1000],
        "default": "1000+",
        "output": {"count": {"$sum": 1}, "avg": {"$avg": "$total"}},
    }},
]

Indexes

# Ensure indexes at startup (idempotent — no-op if already exists)
async def create_indexes(db):
    # Single field
    await db.users.create_index("email", unique=True)

    # Compound — order matters (equality first, range last, sort last)
    await db.orders.create_index([("user_id", 1), ("status", 1), ("created_at", -1)])

    # Text search index
    await db.articles.create_index([("title", "text"), ("body", "text")])

    # TTL — auto-delete documents after expiry_at
    await db.sessions.create_index("expires_at", expireAfterSeconds=0)

    # Sparse — only index docs where field exists
    await db.users.create_index("stripe_customer_id", sparse=True, unique=True)

    # Partial — only index matching docs (smaller index)
    await db.orders.create_index(
        "created_at",
        partialFilterExpression={"status": "active"},
    )

In Agentex, indexes are defined in src/config/mongodb_indexes.py and created on startup automatically.

Transactions (multi-document)

# Requires replica set (or mongos)
async with await client.start_session() as session:
    async with session.start_transaction():
        await db.accounts.update_one(
            {"_id": from_id},
            {"$inc": {"balance": -amount}},
            session=session,
        )
        await db.accounts.update_one(
            {"_id": to_id},
            {"$inc": {"balance": amount}},
            session=session,
        )
        # auto-commits if no exception, auto-aborts on exception

Change Streams (real-time)

# Watch a collection for changes
async def watch_orders():
    pipeline = [{"$match": {"operationType": {"$in": ["insert", "update"]}}}]
    async with db.orders.watch(pipeline) as stream:
        async for change in stream:
            op = change["operationType"]       # "insert", "update", "delete"
            doc = change.get("fullDocument")   # updated document (for insert/update)
            keys = change.get("updateDescription", {}).get("updatedFields", {})
            yield op, doc, keys

Requires replica set. Use change streams to push updates to SSE clients without polling.

Agentex adk.state Pattern

adk.state is MongoDB-backed per-task state storage. Under the hood it's a document per (task_id, agent_id).

from agentex.lib import adk
from project.models import SummarizerState

# Create initial state (insert)
await adk.state.create(
    task_id=task_id,
    agent_id=agent_id,
    data=SummarizerState().model_dump(),
)

# Load state (find_one by task_id + agent_id)
raw = await adk.state.get_by_task_and_agent(task_id=task_id, agent_id=agent_id)
state = SummarizerState(**raw.data)

# Mutate and save (update_one with $set)
state.total_processed += len(batch)
await adk.state.update(
    task_id=task_id,
    agent_id=agent_id,
    data=state.model_dump(),
)

Always load → mutate → save in sequence. Never hold state in workflow memory — Temporal replays will lose it.

Document Design Tips

Query Optimization

EXPLAIN — read the query plan

# winningPlan shows which index was used (or COLLSCAN = no index)
plan = await db.orders.find({"user_id": uid, "status": "active"}).explain()
print(plan["queryPlanner"]["winningPlan"])
# COLLSCAN → add an index
# IXSCAN   → index was used; check "indexName"

# executionStats — actual rows examined vs returned
stats = await db.orders.find({"user_id": uid}).explain("executionStats")
examined = stats["executionStats"]["totalDocsExamined"]
returned = stats["executionStats"]["totalDocsReturned"]
# ratio examined/returned > 10 → index is not selective enough

Covered queries — zero document fetch

A query is covered when the index contains all projected fields — MongoDB never reads the actual document:

# Index: [("user_id", 1), ("status", 1), ("total", 1)]
# Query uses only indexed fields + projects only indexed fields → covered
cursor = db.orders.find(
    {"user_id": uid, "status": "active"},
    {"_id": 0, "user_id": 1, "status": 1, "total": 1},  # only indexed fields
)
# executionStats.totalDocsExamined == 0 confirms it's covered

Index hints

# Force a specific index (useful when the planner picks the wrong one)
cursor = db.orders.find({"user_id": uid}).hint([("user_id", 1), ("created_at", -1)])

# Force collection scan (bypass indexes for small collections)
cursor = db.orders.find({}).hint([("$natural", 1)])

Projection — only fetch what you need

# GOOD: project only needed fields
users = await db.users.find({}, {"name": 1, "email": 1, "_id": 0}).to_list(100)

# BAD: fetch entire document when only name is needed
users = await db.users.find({}).to_list(100)
names = [u["name"] for u in users]

Schema Design Patterns

Embed vs. Reference Decision

# EMBED — order items always loaded with the order
{
    "_id": ObjectId("..."),
    "user_id": ObjectId("..."),
    "total": 149.99,
    "items": [                        # embed: always loaded together
        {"product_id": "p-1", "qty": 2, "price": 49.99},
        {"product_id": "p-2", "qty": 1, "price": 50.01},
    ]
}

# REFERENCE — reviews exist independently; many per product
{
    "_id": ObjectId("..."),
    "product_id": ObjectId("..."),    # reference: independent lifecycle
    "user_id": ObjectId("..."),
    "rating": 4,
    "body": "Great product.",
}

Bucket Pattern — time-series

# BAD: one document per reading → millions of tiny docs, index overhead
{"sensor_id": "s-1", "ts": datetime(...), "temp": 22.4}

# GOOD: one document per hour, readings array inside
{
    "sensor_id": "s-1",
    "hour": datetime(2026, 5, 11, 14, 0, 0, tzinfo=timezone.utc),
    "count": 60,
    "readings": [22.4, 22.5, 22.3, ...],   # one per minute
    "min": 22.3, "max": 22.7, "avg": 22.5, # pre-computed
}
# Index on sensor_id + hour → one index lookup per hour of data

Schema Versioning

# Add schema_version field; migrate lazily on read
async def get_user(user_id: ObjectId) -> dict:
    doc = await db.users.find_one({"_id": user_id})
    version = doc.get("schema_version", 1)
    if version == 1:
        doc = migrate_v1_to_v2(doc)
        await db.users.update_one(
            {"_id": user_id},
            {"$set": {"preferences": doc["preferences"], "schema_version": 2}}
        )
    return doc

TypeScript Patterns (Node.js Driver)

import { MongoClient, ObjectId, type Db } from "mongodb";

const client = new MongoClient("mongodb://localhost:27017", {
  maxPoolSize: 20,
  serverSelectionTimeoutMS: 5000,
});
await client.connect();
const db: Db = client.db("mydb");

// Find one
const user = await db.collection("users").findOne({ email: "alice@example.com" });

// Find many with pagination
const users = await db.collection("users")
  .find({ role: "admin" })
  .sort({ created_at: -1 })
  .skip(page * limit)
  .limit(limit)
  .toArray();

// Insert
const { insertedId } = await db.collection("users").insertOne({
  email: "alice@example.com",
  role: "user",
  created_at: new Date(),
});

// Update
await db.collection("users").updateOne(
  { _id: new ObjectId(id) },
  { $set: { role: "admin", updated_at: new Date() } }
);

// Aggregation
const results = await db.collection("orders").aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$user_id", total: { $sum: "$amount" }, count: { $sum: 1 } } },
  { $sort: { total: -1 } },
  { $limit: 10 },
]).toArray();

// Transaction
const session = client.startSession();
try {
  await session.withTransaction(async () => {
    await db.collection("accounts").updateOne(
      { _id: fromId }, { $inc: { balance: -amount } }, { session }
    );
    await db.collection("accounts").updateOne(
      { _id: toId }, { $inc: { balance: amount } }, { session }
    );
  });
} finally {
  await session.endSession();
}

// Create indexes at startup
await db.collection("users").createIndex({ email: 1 }, { unique: true });
await db.collection("orders").createIndex({ user_id: 1, created_at: -1 });
await db.collection("sessions").createIndex({ expires_at: 1 }, { expireAfterSeconds: 0 });

Red Flags

• No index on query filter or sort fields — MongoDB performs a collection scan for every unindexed query; find({"user_id": x}) on a million-document collection takes seconds without an index on user_id
• Unbounded find() in production — db.collection.find({}) without .limit() loads the entire collection into memory; always add .limit(N) and paginate with a cursor
• $match not as the first pipeline stage — aggregation stages before $match process every document before filtering; placing $match first lets MongoDB use indexes and dramatically reduces the work for subsequent stages
• datetime.utcnow() instead of datetime.now(timezone.utc) — utcnow() returns a naive datetime with no timezone info and is deprecated in Python 3.12; use datetime.now(timezone.utc) to get a timezone-aware UTC datetime that Motor stores correctly
• Transactions without a replica set — client.start_session() multi-document transactions require a replica set (or mongos); on a standalone instance they raise a server error; use a replica set even in development (mongo --replSet rs0)
• Embedding unbounded arrays — pushing to a tags[] or events[] array without a $slice limit grows the document indefinitely, eventually hitting the 16 MB BSON document size limit; cap arrays at creation time using $push with $slice
• Holding state in workflow memory instead of adk.state — Temporal replays recreate the workflow from scratch; any in-memory state not persisted to MongoDB via adk.state is lost on replay, causing the workflow to behave differently than the first execution

Checklist

• Indexes created at startup for all query filter and sort fields
• Compound indexes: equality fields first, range/sort fields last
• TTL index used for session/temp data instead of manual cleanup
• find() always has .limit() — never unbounded cursor in production
• Aggregation $match is the first stage (filters before loading docs)
• $lookup result arrays have $limit or $slice if sub-docs can be large
• Transactions used for multi-collection writes that must be atomic
• adk.state always follows load → mutate → save pattern in Temporal activities
• datetime.now(timezone.utc) used (not datetime.utcnow() — deprecated)