Introduction
Python is one of the most accessible languages for building applications backed by MongoDB, yet many teams struggle to choose between synchronous and asynchronous drivers, or end up with inconsistent data access patterns across their codebase. MongoDB with Python: a practical MongoDB & NoSQL guide will walk you through the real-world choices we make when integrating MongoDB into FastAPI services, batch jobs, and microservices at scale.
By the end of this article, you'll understand when to reach for PyMongo, when async patterns with Motor or PyMongo's async support make sense, and how Beanie ODM with Pydantic models can reduce boilerplate while keeping your schema intentional.
PyMongo for Synchronous Code
PyMongo is MongoDB's official synchronous Python driver. It's battle-tested, straightforward, and the natural choice for scripts, traditional Flask applications, and blocking I/O workloads where you aren't juggling hundreds of concurrent connections.
In production systems handling order processing, data exports, or scheduled aggregation jobs, we reach for PyMongo when the work is naturally sequential. A bulk import job that reads from a CSV and writes to MongoDB doesn't need async complexity; it needs clarity and debuggability.
Here's a realistic example: a worker that processes invoices and records them in MongoDB.
from pymongo import MongoClient, ASCENDING
from pymongo.errors import DuplicateKeyError
import logging
logger = logging.getLogger(__name__)
class InvoiceRepository:
def __init__(self, connection_string: str):
self.client = MongoClient(connection_string)
self.db = self.client["billing"]
self.invoices = self.db["invoices"]
def process_invoice(self, invoice_data: dict) -> dict:
"""Insert or update an invoice. Raises DuplicateKeyError if unique constraint violated."""
try:
result = self.invoices.insert_one(invoice_data)
logger.info(f"Invoice inserted: {result.inserted_id}")
return {"id": str(result.inserted_id), "status": "created"}
except DuplicateKeyError:
logger.warning(f"Invoice already exists: {invoice_data.get('invoice_number')}")
raise
def fetch_invoices_by_customer(self, customer_id: str, limit: int = 100):
"""Retrieve invoices for a customer with efficient pagination."""
return list(
self.invoices.find(
{"customer_id": customer_id},
{"created_at": -1}
).limit(limit)
)
def close_connection(self):
self.client.close()
# Usage in a batch job
if __name__ == "__main__":
repo = InvoiceRepository("mongodb://localhost:27017")
invoices = [
{"invoice_number": "INV-001", "customer_id": "cust_123", "amount": 150.00},
{"invoice_number": "INV-002", "customer_id": "cust_456", "amount": 275.50},
]
for invoice in invoices:
try:
repo.process_invoice(invoice)
except DuplicateKeyError:
print(f"Skipping duplicate invoice: {invoice['invoice_number']}")
repo.close_connection()PyMongo's MongoClient maintains a connection pool by default, so creating a single instance and reusing it across your application is the pattern we follow. Keep the client alive as long as your process runs, and close it gracefully on shutdown.
Motor and PyMongo Async for FastAPI
When you're building a FastAPI service handling hundreds of concurrent requests, blocking on database calls defeats the entire purpose of async. This is where Motor (an async MongoDB driver built on PyMongo) or PyMongo's native async support come in.
Motor is the traditional choice and remains solid for production use. More recently, PyMongo added async support directly, giving teams another option. Both integrate cleanly with FastAPI's dependency injection.
Here's a FastAPI endpoint that retrieves user data asynchronously:
from fastapi import FastAPI, Depends, HTTPException
from motor.motor_asyncio import AsyncClient, AsyncDatabase
from contextlib import asynccontextmanager
import os
app = FastAPI()
# Global Motor client
motor_client: AsyncClient = None
db: AsyncDatabase = None
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
global motor_client, db
motor_client = AsyncClient(os.getenv("MONGODB_URI", "mongodb://localhost:27017"))
db = motor_client["users_db"]
yield
# Shutdown
motor_client.close()
app = FastAPI(lifespan=lifespan)
async def get_db():
return db
@app.get("/users/{user_id}")
async def get_user(user_id: str, database: AsyncDatabase = Depends(get_db)):
"""Fetch a user by ID asynchronously."""
from bson import ObjectId
try:
user = await database["users"].find_one({"_id": ObjectId(user_id)})
if not user:
raise HTTPException(status_code=404, detail="User not found")
# Convert ObjectId to string for JSON serialization
user["_id"] = str(user["_id"])
return user
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/users")
async def create_user(name: str, email: str, database: AsyncDatabase = Depends(get_db)):
"""Create a new user asynchronously."""
result = await database["users"].insert_one({"name": name, "email": email})
return {"id": str(result.inserted_id), "status": "created"}Notice how the lifespan context manager handles Motor client initialization and cleanup. This pattern ensures the client is ready before any request hits your handlers and is properly closed on shutdown. Every endpoint awaits database operations, keeping the event loop free for other requests.
Motor's API mirrors PyMongo's, so transitioning between them is low-friction. If you're already comfortable with PyMongo, Motor feels natural.
Beanie ODM with Pydantic Models
As your MongoDB application grows, raw document handling becomes tedious and error-prone. Beanie is an async-first ODM (Object-Document Mapper) that binds Pydantic models directly to MongoDB collections, giving you validation, serialization, and type safety without the overhead of traditional ORMs.
Beanie is particularly valuable when you need strong schema contracts. Instead of relying on informal documentation about what fields exist and their types, Pydantic models are your source of truth.
Here's a realistic example using Beanie in a SaaS product API:
from beanie import Document, init_beanie, Indexed
from pydantic import BaseModel, Field, EmailStr
from motor.motor_asyncio import AsyncClient
from typing import Optional
from datetime import datetime
from enum import Enum
class SubscriptionTier(str, Enum):
FREE = "free"
PRO = "pro"
ENTERPRISE = "enterprise"
class UserProfile(BaseModel):
"""Nested model for user metadata."""
bio: Optional[str] = None
avatar_url: Optional[str] = None
company: Optional[str] = None
class User(Document):
"""User document with Pydantic validation and Beanie ORM features."""
email: Indexed(EmailStr, unique=True) # Indexed and unique
name: str
profile: UserProfile = Field(default_factory=UserProfile)
subscription_tier: SubscriptionTier = SubscriptionTier.FREE
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)
is_active: bool = True
class Settings:
name = "users" # Collection name
async def init_db():
"""Initialize Beanie with the MongoDB connection."""
client = AsyncClient("mongodb://localhost:27017")
await init_beanie(
database=client["myapp_db"],
models=[User]
)
async def create_user(email: str, name: str) -> User:
"""Create and save a user document."""
user = User(email=email, name=name)
await user.insert() # Beanie handles insertion
return user
async def find_user_by_email(email: str) -> Optional[User]:
"""Query users with type safety."""
return await User.find_one(User.email == email)
async def update_user_profile(user_id: str, bio: str, avatar_url: str):
"""Update nested fields elegantly."""
user = await User.get(user_id)
if user:
user.profile.bio = bio
user.profile.avatar_url = avatar_url
user.updated_at = datetime.utcnow()
await user.save()
return user
async def find_pro_users():
"""Query with multiple filters."""
return await User.find(
User.subscription_tier == SubscriptionTier.PRO,
User.is_active == True
).to_list()Beanie's approach is elegant: your Pydantic models are also your documents. The Indexed type enforces index creation, validation happens automatically on save, and queries use Python expressions instead of raw MongoDB filter dictionaries.
One key advantage: Beanie integrates seamlessly with FastAPI. When you return a Beanie document from an endpoint, Pydantic handles serialization to JSON automatically, including proper ObjectId conversion.
Create Indexes at Startup
Indexes are critical for query performance, but they're often an afterthought. In production, you can't afford to discover missing indexes in production traffic. The safest approach is to define and create indexes as part of your application startup.
With PyMongo, you explicitly create indexes:
from pymongo import MongoClient, ASCENDING, DESCENDING
def ensure_indexes(db):
"""Create all required indexes on application startup."""
# Single field index on email
db["users"].create_index([("email", ASCENDING)], unique=True, sparse=True)
# Compound index for efficient filtering and sorting
db["orders"].create_index([
("customer_id", ASCENDING),
("created_at", DESCENDING)
])
# Sparse index for optional fields
db["logs"].create_index([("session_id", ASCENDING)], sparse=True)
print("Indexes created successfully")
# In your main setup
if __name__ == "__main__":
client = MongoClient("mongodb://localhost:27017")
db = client["myapp"]
ensure_indexes(db)With Beanie, index creation is declarative and automatic:
from beanie import Document, Indexed
from typing import Optional
class Order(Document):
customer_id: Indexed(str) # Creates a simple index
email: Indexed(str, unique=True) # Creates a unique index
created_at: Indexed(datetime)
amount: float
class Settings:
name = "orders"
indexes = [
# Compound index
[("customer_id", 1), ("created_at", -1)],
]
# Beanie creates these indexes automatically when init_beanie is calledThe key pattern: always create indexes in code during startup, not manually via MongoDB shell. This ensures they exist before your application serves traffic and makes index definitions reviewable in version control.
Common Mistakes
- Creating a new MongoClient for every request: Connection pools are expensive to initialize. Reuse a single client instance across your application lifetime. In FastAPI, use lifespan events to manage the client.
- Forgetting to close connections: Unclosed connections leak resources. Always close your client gracefully on shutdown, whether using context managers or explicit close calls.
- Mixing sync and async code without caution: Don't call blocking PyMongo methods in an async function. If you need both sync and async, use PyMongo in sync contexts and Motor in async contexts.
- Assuming Beanie will auto-migrate your schema: Beanie validates on insert/update, but it won't backfill existing documents or remove fields automatically. Plan migrations explicitly.
- Skipping index creation: Running queries without appropriate indexes will be slow. Define indexes upfront and verify they're created on application startup.
- Returning raw Beanie/PyMongo documents from FastAPI endpoints without serialization: ObjectId fields won't serialize to JSON by default. Explicitly convert them to strings or use Pydantic's Field configurations.
Key Takeaways
- Use PyMongo for synchronous, sequential workloads like batch jobs, scripts, and traditional blocking code paths.
- Use Motor or PyMongo async in FastAPI and other async frameworks to handle concurrent requests without blocking the event loop.
- Beanie ODM reduces boilerplate and provides type safety when combined with Pydantic, making it ideal for APIs with strong schema contracts.
- Always create indexes at startup, either explicitly with PyMongo or declaratively with Beanie, and verify they exist before serving traffic.
- Manage MongoDB client lifecycle carefully: initialize once, reuse across requests, and close gracefully on shutdown using lifespan contexts or context managers.
- Be intentional about serialization when returning MongoDB documents from web APIs; convert ObjectId to string and validate nested structures with Pydantic.
Comments
Loading comments…