Python Driver (PyMongo) — MongoDB

PyMongo — The Python Driver

PyMongo is the official MongoDB driver for Python. It maps documents to plain Python dictionaries, and gives you the same connection pooling, cursor, and transaction primitives as every other official driver.

Connecting

Python

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017", maxPoolSize=20)
db = client.shop            # attribute access
users = db["users"]         # or dict-style access

# Verify the connection
client.admin.command("ping")

CRUD Operations

Python

# Create
users.insert_one({"email": "alice@example.com", "name": "Alice"})
users.insert_many([{"name": "Bob"}, {"name": "Carol"}])

# Read
alice = users.find_one({"email": "alice@example.com"})
admins = list(users.find({"role": "admin"}).sort("createdAt", -1).limit(20))

# Update
users.update_one({"_id": alice["_id"]}, {"$set": {"name": "Alice S."}})
users.update_many({"role": "user"}, {"$set": {"active": True}})

# Delete
users.delete_one({"_id": alice["_id"]})
users.delete_many({"active": False})

Cursors

find() returns a lazy Cursor — iterate it directly instead of materializing a list, unless the result set is small and known.

Python

cursor = users.find({"role": "user"}, {"email": 1, "name": 1})

for doc in cursor:
    print(doc["email"])

# batch_size controls how many documents are fetched per round trip
cursor = users.find().batch_size(500)

Aggregation

Python

pipeline = [
    {"$match": {"status": "shipped"}},
    {"$group": {"_id": "$customerId", "total": {"$sum": "$amount"}}},
    {"$sort": {"total": -1}},
    {"$limit": 10}
]

for doc in db.orders.aggregate(pipeline):
    print(doc)

Indexes

Python

users.create_index("email", unique=True)
users.create_index([("lastName", 1), ("firstName", 1)])
users.create_index(
    "email",
    unique=True,
    partialFilterExpression={"deletedAt": {"$exists": False}}
)

for idx in users.list_indexes():
    print(idx)

Error Handling

Python

from pymongo.errors import DuplicateKeyError, PyMongoError

try:
    users.insert_one({"email": "alice@example.com"})
except DuplicateKeyError:
    print("Email already exists")
except PyMongoError as exc:
    print("MongoDB error:", exc)

BSON Types in Python

PyMongo automatically converts between BSON types and native Python types. A few need attention: ObjectId and timezone-aware datetime.

Python

from bson import ObjectId
from datetime import datetime, timezone

doc = users.find_one({"_id": ObjectId("507f1f77bcf86cd799439011")})

users.insert_one({
    "name": "Dana",
    "createdAt": datetime.now(timezone.utc)   # store as UTC — MongoDB stores UTC internally
})

# ObjectId does NOT serialize to plain JSON — convert explicitly at API boundaries
str(doc["_id"])

Warning

MongoDB stores all Date values internally as UTC. If you insert a naive (timezone-unaware) Python datetime, PyMongo assumes it's already UTC — always use datetime.now(timezone.utc) to avoid silent offset bugs.

Motor — Async PyMongo

For asyncio-based applications (FastAPI, aiohttp), use Motor, the official async wrapper around the same driver internals — same API shape, but every call is awaited.

Python

import motor.motor_asyncio

client = motor.motor_asyncio.AsyncIOMotorClient("mongodb://localhost:27017")
db = client.shop

async def get_user(email: str):
    return await db.users.find_one({"email": email})

PyMongo Quick Reference

Task	Call
Connect	`MongoClient(uri)`
Insert one / many	`insert_one()` / `insert_many()`
Find one / many	`find_one()` / `find()`
Update one / many	`update_one()` / `update_many()`
Replace a document	`replace_one()`
Delete one / many	`delete_one()` / `delete_many()`
Aggregate	`aggregate(pipeline)`
Create an index	`create_index()`
Count matching documents	`count_documents(filter)`

Tip

estimated_document_count() is much faster than count_documents() for a rough total, because it reads collection metadata instead of scanning — use it for dashboards where an approximate count is fine.

Summary

PyMongo mirrors the same core concepts as the Node.js driver: MongoClient pooling, lazy cursors, and session-based transactions.
Always store datetimes as timezone-aware UTC to avoid offset bugs.
ObjectId needs explicit str() conversion before JSON serialization.
Reach for Motor when building on asyncio (FastAPI, aiohttp) instead of blocking PyMongo calls.