PyMongo Cheat Sheet

1Install & connect4.17

pip install pymongo # or "pymongo[srv]" for Atlas★
The srv extra enables mongodb+srv:// connection strings used by MongoDB Atlas.
from pymongo import MongoClient client = MongoClient("mongodb://localhost:27017/")★
One client per app — it's a connection pool, thread-safe, and connects lazily on first operation. Atlas: MongoClient("mongodb+srv://user:pass@cluster.mongodb.net/").
client.admin.command("ping") # verify connectivity
Since connection is lazy, ping to fail fast at startup if the server is unreachable.
client.close() # at shutdown; don't create one per request
Reuse the single client for the process lifetime. Creating clients per request exhausts connections.

2Collections & insertdatabases are dicts of dicts

db = client["shop"] # or client.shop users = db["users"] # a collection★
Databases and collections are created lazily on first write — no CREATE needed.
res = users.insert_one({"name": "Ada", "age": 36}) res.inserted_id # ObjectId(...)★
Documents are plain dicts. If you omit _id, MongoDB assigns an ObjectId. Nested dicts/lists are fine.
res = users.insert_many([{"name": "Al"}, {"name": "Grace"}]) res.inserted_ids★
Batch insert. ordered=False keeps going past a failing doc (e.g. duplicate key).
from bson import ObjectId users.find_one({"_id": ObjectId("665f1c...")})
_id is an ObjectId, not a string — wrap a hex id from a URL/param in ObjectId() to query by it.

3Query with finddict filters + $-operators

users.find_one({"name": "Ada"}) # one dict or None★
Exact match on a field. An empty filter {} matches everything.
for doc in users.find({"age": {"$gt": 30}}): ... # find() returns a lazy cursor★
Operators: $gt $gte $lt $lte $ne $in $nin $and $or $exists $regex. The cursor streams — iterate once.
users.find({}, {"name": 1, "_id": 0}) \ .sort("age", -1).limit(10).skip(20)★
Projection (2nd arg): 1 include / 0 exclude. Chain sort/limit/skip on the cursor.
users.count_documents({"age": {"$gte": 18}})
Exact count for a filter. estimated_document_count() is fast but approximate (whole collection only).

4Update & deleteupdate operators

users.update_one({"name": "Ada"}, {"$set": {"age": 37}})★
Use an update operator ($set) — the 2nd arg is a change spec, not a replacement doc. Matches the first matching document.
users.update_many({"active": True}, {"$inc": {"visits": 1}})★
Operators: $set $unset $inc $mul $push $pull $addToSet $rename $min $max. update_many hits all matches.
users.update_one({"email": e}, {"$set": {...}}, upsert=True)★
upsert=True inserts the document if no match exists — the classic get-or-create.
users.delete_one({"_id": oid}) · users.delete_many({"active": False})★
Delete first match / all matches. find_one_and_update(..., return_document=ReturnDocument.AFTER) updates and returns the doc atomically.

5Aggregation pipelinetransform & group

users.aggregate([ {"$match": {"active": True}}, {"$group": {"_id": "$city", "n": {"$sum": 1}}}, {"$sort": {"n": -1}}, ])★
A list of stages, each a dict. Data flows stage→stage. $group._id is the group key; $sum:1 counts.
# stages: $match $project $group $sort $limit $skip $unwind
$project reshapes fields, $unwind explodes an array into one doc per element.
{"$lookup": {"from": "orders", "localField": "_id", "foreignField": "user_id", "as": "orders"}}
$lookup is a join — embeds matching docs from another collection as an array.

6Indexes & bulkspeed + batched writes

users.create_index("email", unique=True)★
Index the fields you filter/sort on — without one, queries do a full collection scan. unique=True enforces uniqueness.
from pymongo import ASCENDING, DESCENDING users.create_index([("city", ASCENDING), ("age", DESCENDING)])
Compound index — order matters and should match your query/sort pattern. Also text and 2dsphere (geo) indexes.
from pymongo import InsertOne, UpdateOne, DeleteOne users.bulk_write([InsertOne({...}), UpdateOne({"_id": 1}, {"$set": {...}})], ordered=False)★
Batch mixed writes in one round-trip. ordered=False parallelizes and continues past errors.

7Transactionsmulti-doc atomicity

with client.start_session() as s: s.with_transaction(lambda s: transfer(s))★
The recommended pattern: with_transaction runs your callback and handles commit + automatic retries on transient errors.
def transfer(s): accts.update_one({"_id": a}, {"$inc": {"bal": -100}}, session=s) accts.update_one({"_id": b}, {"$inc": {"bal": 100}}, session=s)★
Pass session=s to every operation in the transaction — operations without it aren't part of it.
# transactions REQUIRE a replica set (or Atlas)
A standalone mongod can't run transactions. Single-document writes are always atomic without one.

8AsyncAsyncMongoClient (supersedes Motor)

from pymongo import AsyncMongoClient client = AsyncMongoClient("mongodb://localhost:27017/") await client.shop.users.insert_one({"name": "Ada"})★
Native async (stable since 4.13) — the same API, awaited. Use this instead of the retiring Motor driver.
doc = await client.shop.users.find_one({"name": "Ada"}) async for d in client.shop.users.find({"age": {"$gt": 30}}): ...★
await single-result calls; async for over a cursor. aggregate, bulk_write, sessions all have awaitable forms.
await client.aclose()
Close the async client on shutdown (aclose, the async counterpart of close).

!Common gotchasread before shipping

update needs $-operators
update_one(f, {"age": 37}) is wrong — a bare dict without operators errors (or, historically, replaced the whole doc). Use {"$set": {...}}.
_id is an ObjectId, not a str
Querying {"_id": "665f..."} won't match. Wrap it: ObjectId(hex). Serialize ids to str for JSON APIs.
find() returns a one-shot cursor
It's lazy and exhausted after one iteration. Re-query, or materialize with list(cursor) if you need it twice.
no schema = silent typos
Misspelling a field on insert/update just creates a new field — nothing errors. Consider JSON-schema validation on the collection.
transactions need a replica set
They fail on a standalone server. For local dev, run a single-node replica set or use Atlas.
index or full scan
Unindexed filters/sorts scan every document. create_index the fields you query; check with .explain().
one MongoClient per process
It's a pool — share it. Creating one per request (or per function) leaks connections and is slow.

PyMongo cheat sheet