MongoDB with Python: a Python Backend guide for developers

MongoDB with Python: a Python Backend guide for developers

Building scalable backends often means choosing a document database that plays well with Python's async ecosystem. MongoDB with Python gives teams flexibility in schema design and horizontal scaling, but only when paired with the right driver and architectural choices. In this guide, we'll walk through PyMongo and Motor, show you how to model documents properly, optimize with indexes, and know when transactions actually matter.

Why MongoDB fits modern Python backends

MongoDB's document model aligns naturally with Python dictionaries and JSON structures, making it feel intuitive to developers coming from REST APIs and FastAPI applications. Unlike relational databases, MongoDB lets you evolve your schema without migrations, which speeds up iteration during early development. However, this flexibility comes with responsibility—you still need to think about data modeling, indexing, and query patterns just as carefully as you would with PostgreSQL.

The Python ecosystem offers two main drivers: PyMongo for synchronous code and Motor for async applications. Most modern backend services we build today use Motor, especially when paired with FastAPI, because it prevents blocking I/O from stalling your event loop.

Motor for async FastAPI apps

Motor is an async MongoDB driver that integrates seamlessly with FastAPI's async runtime. Unlike PyMongo, which blocks during database operations, Motor uses coroutines so your application can handle thousands of concurrent requests without thread pool exhaustion.

Here's a typical FastAPI + Motor setup:

from motor.motor_asyncio import AsyncClient, AsyncDatabase
from fastapi import FastAPI
from contextlib import asynccontextmanager

app = FastAPI()
mongo_client: AsyncClient = None
db: AsyncDatabase = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global mongo_client, db
    mongo_client = AsyncClient("mongodb://localhost:27017")
    db = mongo_client["myapp_db"]
    yield
    mongo_client.close()

app = FastAPI(lifespan=lifespan)

@app.get("/users/{user_id}")
async def get_user(user_id: str):
    user = await db.users.find_one({"_id": ObjectId(user_id)})
    return user

Notice the lifespan context manager—this ensures Motor connects once at startup and closes cleanly on shutdown. Storing db globally works because Motor handles connection pooling internally; you don't create a new connection per request.

Motor's methods mirror PyMongo's API but are all async. Common operations become natural:

async def create_order(order_data: dict):
    result = await db.orders.insert_one(order_data)
    return result.inserted_id

async def update_user_email(user_id: str, new_email: str):
    await db.users.update_one(
        {"_id": ObjectId(user_id)},
        {"$set": {"email": new_email}}
    )

async def fetch_recent_orders(limit: int = 10):
    cursor = db.orders.find({"status": "completed"}).sort("created_at", -1).limit(limit)
    return await cursor.to_list(length=None)

A critical detail: when using Motor, always use await with database operations. Forgetting await on a Motor coroutine returns a coroutine object instead of actual data, causing silent bugs that only surface at runtime.

Model documents with Pydantic

While MongoDB's schemaless nature is powerful, leaving your data model entirely implicit leads to bugs. Pydantic bridges this gap by validating and serializing documents consistently across your application.

Define a Pydantic model for each main document type:

from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List
from datetime import datetime
from bson import ObjectId

class User(BaseModel):
    id: Optional[ObjectId] = Field(None, alias="_id")
    email: EmailStr
    name: str
    created_at: datetime = Field(default_factory=datetime.utcnow)
    roles: List[str] = ["user"]
    metadata: dict = {}

    class Config:
        populate_by_name = True
        json_encoders = {
            ObjectId: str
        }

class Order(BaseModel):
    id: Optional[ObjectId] = Field(None, alias="_id")
    user_id: ObjectId
    items: List[dict]
    total: float
    status: str = "pending"
    created_at: datetime = Field(default_factory=datetime.utcnow)

    class Config:
        populate_by_name = True
        json_encoders = {
            ObjectId: str
        }

Use the alias="_id" pattern because MongoDB uses _id as the primary key, but Python naming conventions prefer lowercase without underscores. The populate_by_name = True config allows Pydantic to accept both _id and id interchangeably.

When inserting or retrieving documents, convert between Pydantic models and dictionaries:

async def create_user(user: User):
    user_dict = user.dict(by_alias=True, exclude_none=True)
    result = await db.users.insert_one(user_dict)
    return result.inserted_id

async def get_user_by_email(email: str) -> Optional[User]:
    doc = await db.users.find_one({"email": email})
    if not doc:
        return None
    return User(**doc)

@app.post("/users")
async def register_user(user: User):
    user_id = await create_user(user)
    return {"user_id": str(user_id)}

Pydantic validates the email format, ensures all required fields are present, and handles type coercion. This prevents invalid documents from ever reaching MongoDB, catching problems early in your application layer.

Create indexes at startup

Indexes are invisible to your code but dramatically impact query performance. A missing index turns a millisecond query into a full collection scan that touches every document. The best time to create indexes is during application startup, not in production via ad-hoc commands.

Define indexes declaratively and ensure they exist when your app boots:

async def setup_indexes():
    # Single-field indexes
    await db.users.create_index("email", unique=True, sparse=True)
    await db.users.create_index("created_at")
    
    # Compound indexes for common queries
    await db.orders.create_index([("user_id", 1), ("created_at", -1)])
    await db.orders.create_index([("status", 1), ("created_at", -1)])
    
    # Text index for search
    await db.posts.create_index([("title", "text"), ("body", "text")])

@asynccontextmanager
async def lifespan(app: FastAPI):
    global mongo_client, db
    mongo_client = AsyncClient("mongodb://localhost:27017")
    db = mongo_client["myapp_db"]
    await setup_indexes()
    yield
    mongo_client.close()

MongoDB's create_index is idempotent—calling it multiple times with the same parameters is safe. This means you can call it every startup without worry.

Think about your query patterns first. If you frequently filter orders by user_id and sort by created_at, create that compound index. If you search user profiles by email, make email unique and sparse (sparse means null values don't count toward uniqueness, letting multiple documents omit the field).

Monitor index usage in production. MongoDB provides index statistics that show which indexes are actually being used. Unused indexes waste memory and slow down writes, so remove them.

Use transactions only when needed

ACID transactions in MongoDB are powerful but come with real costs. They acquire locks, add network latency, and complicate your code. Many teams reach for transactions reflexively, but most business logic doesn't need them.

Transactions shine when you need to update multiple documents atomically and a partial failure would corrupt your data. A common example is transferring money between accounts—you must debit one and credit the other in a single atomic operation, or you lose consistency.

Here's a transaction-based transfer:

async def transfer_funds(from_user_id: str, to_user_id: str, amount: float):
    async with await mongo_client.start_session() as session:
        async with session.start_transaction():
            await db.accounts.update_one(
                {"_id": ObjectId(from_user_id)},
                {"$inc": {"balance": -amount}},
                session=session
            )
            await db.accounts.update_one(
                {"_id": ObjectId(to_user_id)},
                {"$inc": {"balance": amount}},
                session=session
            )
            # If an exception occurs here, the entire transaction rolls back

But most operations don't need transactions. Creating an order and logging it to analytics? Those are independent operations—let them fail independently and use retry logic if needed. Incrementing a view counter? Atomicity at the document level is enough; MongoDB's $inc operator is atomic by itself.

Overusing transactions signals a design problem. If you're nesting transactions or holding them open across network calls, redesign your data model or business logic instead. Transactions in MongoDB block other operations on the same documents, so they're a finite resource you should guard carefully.

Common mistakes

Forgetting to await Motor operations: This is the most common bug. result = db.users.find_one(...) without await leaves you with a coroutine object, not actual data. Linters can catch this if configured correctly.

Creating new connections per request: Some teams create a new MongoDB client for each API endpoint. This exhausts connection pools and kills performance. Initialize the client once at startup and reuse it globally.

No indexes in development: Queries run fast against small test datasets without indexes, so developers miss indexing during development. Then production queries are slow. Always create indexes in your startup logic, even in dev environments.

Storing deeply nested documents: MongoDB allows arbitrary nesting, but deeply nested structures become hard to query and update. Keep your documents relatively flat or use subdocuments only where they're truly atomic units.

Misusing transactions for performance: Transactions are slower than non-transactional operations. If you're using transactions to "batch" multiple operations for speed, you're doing it wrong. Use bulk writes instead.

Key takeaways

• Motor is the async MongoDB driver for FastAPI; PyMongo blocks and is unsuitable for high-concurrency services.
• Initialize Motor's client once at application startup using a lifespan context manager and store the db reference globally.
• Model your documents with Pydantic to validate structure and handle ObjectId serialization consistently.
• Create indexes during startup with a setup function called from your lifespan handler; index design follows from query patterns, not guesses.
• Transactions are necessary for multi-document atomicity but carry performance costs; use them only when partial updates would corrupt your data.
• Monitor your indexes in production and remove unused ones to preserve memory and write performance.