PyMongo Basics

Introduction

PyMongo is the official synchronous Python driver for MongoDB. It maps Python dicts to BSON documents and exposes insert_one, find, update_one, and bulk_write with the same semantics as mongosh. This chapter connects from a virtual environment, performs CRUD, handles ObjectId, errors, and transactions—the foundation for FastAPI integration.

Prerequisites

Install

bash
python -m venv .venv
source .venv/bin/activate
pip install "pymongo[srv]>=4.6"

[srv] adds dnspython for mongodb+srv:// Atlas URIs.

requirements.txt:

text
pymongo[srv]==4.10.1

Connect With MongoClient

mongo_demo.py:

Code explanation:

  • One MongoClient per process—thread-safe pool inside
  • client["hello_demo"] selects database
  • db["posts"] selects collection (same as db.posts)

Close on shutdown (CLI scripts):

python
client.close()

Long-running apps: create client at startup, close at exit.

Insert

Find

Code explanation:

  • find returns cursor—lazy until iteration
  • Projection: 1 include, 0 exclude
  • sort("createdAt", -1) descending

Update and Delete

bulk_write

python
from pymongo import InsertOne, UpdateOne, DeleteOne
 
posts.bulk_write([
    InsertOne({"title": "Bulk", "slug": "bulk", "published": True}),
    UpdateOne({"slug": "pymongo-intro"}, {"$inc": {"viewCount": 1}}),
    DeleteOne({"slug": "bulk"}),
], ordered=False)

ObjectId and JSON APIs

ObjectId is not JSON-serializable by default:

python
from bson import ObjectId
import json
 
def default_serializer(obj):
    if isinstance(obj, ObjectId):
        return str(obj)
    raise TypeError(f"Type {type(obj)} not serializable")
 
doc = posts.find_one({"slug": "pymongo-intro"})
print(json.dumps(doc, default=default_serializer))

FastAPI/Pydantic use str for id in response models — chapter 20.

Parse string id from URL:

python
def parse_object_id(value: str) -> ObjectId:
    if not ObjectId.is_valid(value):
        raise ValueError("Invalid id")
    return ObjectId(value)

Error Handling

python
from pymongo.errors import DuplicateKeyError, OperationFailure
 
try:
    posts.insert_one({"_id": "fixed-id", "title": "Once"})
    posts.insert_one({"_id": "fixed-id", "title": "Twice"})
except DuplicateKeyError:
    print("Duplicate _id or unique index")
 
try:
    client.admin.command("nonexistentCommand")
except OperationFailure as exc:
    print(exc.code, exc.details)

Retry transient errors with exponential backoff in workers—not blind retry on DuplicateKeyError.

Transactions (Replica Set)

Requires replica set URI — chapter 14.

Indexes From PyMongo

python
posts.create_index("slug", unique=True)
posts.create_index([("published", 1), ("createdAt", -1)])
print(posts.index_information())

Project Layout Hint

text
app/
├── db/
│   ├── client.py      # MongoClient singleton
│   └── collections.py # get_posts_collection()
├── repositories/
│   └── posts.py       # CRUD functions
└── main.py

Keeps route handlers thin—similar to SQLAlchemy repositories in FastAPI SQLAlchemy.

Post-Chapter Checklist

  • MongoClient connects and ping succeeds
  • CRUD via insert_one, find, update_one, delete_one
  • ObjectId converted for JSON
  • DuplicateKeyError handled for unique slug

FAQ

PyMongo vs Motor?

PyMongo is sync; Motor wraps it for async/await — Motor chapter. FastAPI sync routes can use PyMongo directly.

ODM like MongoEngine?

ODM adds schema classes; this track uses PyMongo + Pydantic for clarity.

Django MongoDB?

Use MongoDB extension or separate service; Django ORM is SQL-first — Django track.

Connection string in code?

Use os.environ["MONGODB_URI"] — Security.

Type hints?

from pymongo.collection import Collection for Collection[dict] (3.9+ generic) optional.