MongoDB CRUD Operations with Python (PyMongo)

PyMongo is the official, MongoDB-maintained driver for Python. It gives you everything you need to run CRUD operations — Create, Read, Update and Delete — against a MongoDB collection straight from Python.

This guide is updated for PyMongo 4.x (the current major release). The short version:

• Create — insert_one(), insert_many()
• Read — find(), find_one(), count_documents()
• Update — update_one(), update_many(), replace_one(), find_one_and_update()
• Delete — delete_one(), delete_many()

If you are migrating from PyMongo 3.x, note that the legacy insert(), update(), remove(), save(), find_and_modify() and count() helpers were removed in PyMongo 4.0 — every example below uses only the supported modern API.

Install PyMongo

Install the driver with pip. Add the srv extra when you connect to MongoDB Atlas (or any mongodb+srv:// URI), which pulls in the DNS support needed for SRV connection strings:

# Standard install
pip install pymongo

# With SRV / Atlas connection-string support
pip install "pymongo[srv]"

Connect to MongoDB

Create a single MongoClient and reuse it for the life of your application — it manages an internal connection pool, so spinning up a new client per request is wasteful. Never hard-code credentials; read the connection string from an environment variable.

import os
from pymongo import MongoClient

# Local MongoDB
client = MongoClient("mongodb://localhost:27017")

# MongoDB Atlas (SRV connection string) — read from an env var, never hard-code
client = MongoClient(os.environ["MONGODB_URI"])
# e.g. mongodb+srv://<user>:<password>@cluster0.abcd.mongodb.net/?retryWrites=true&w=majority

db = client["test"]      # select the database
col = db["person"]       # select the collection (created lazily on first write)

You can also use MongoClient as a context manager so the connection pool is closed cleanly when you are done:

with MongoClient(os.environ["MONGODB_URI"]) as client:
    col = client["test"]["person"]
    col.insert_one({"name": "John", "salary": 100})
# client is closed automatically on exit

# If you create the client manually, close it yourself on shutdown:
# client.close()

Create — insert_one() and insert_many()

MongoDB stores each record as a BSON document. If the target collection does not exist yet, the first insert creates it. Documents are plain Python dicts — note the keys must be strings ("name", not name).

insert_one() adds a single document and returns an InsertOneResult; insert_many() adds a list and returns an InsertManyResult. Both expose the generated _id value(s) — an ObjectId when you don't supply one.

# Insert a single document
result = col.insert_one({"name": "John", "salary": 100})
print(result.inserted_id)        # ObjectId('57611d4b1aa303032ad5ba9e')

# Insert many documents at once
result = col.insert_many([
    {"name": "George", "salary": 100},
    {"name": "Steve",  "salary": 120},
    {"name": "David",  "salary": 140},
])
print(result.inserted_ids)       # [ObjectId('...'), ObjectId('...'), ObjectId('...')]

Read — find(), find_one() and count_documents()

find_one() returns the first matching document (a dict) or None. find() returns a lazy cursor you iterate over. Pass a filter dict to narrow results and a projection to control which fields come back. Use count_documents() to count matches — the old count() was removed in 4.0.

# One document (or None)
col.find_one({"name": "John"})
# {'_id': ObjectId('57611a71...'), 'name': 'John', 'salary': 100}

# Iterate a cursor
for doc in col.find({"salary": {"$gte": 100}}):
    print(doc["name"])

# Projection: include name, exclude _id
col.find_one({"name": "John"}, {"_id": 0, "name": 1})
# {'name': 'John'}

# Sort, skip and limit (pagination)
cursor = col.find().sort("salary", -1).skip(10).limit(5)

# Count matching documents (replaces the removed count())
col.count_documents({"salary": {"$gte": 100}})

Update — update_one(), update_many() and replace_one()

Updates take a filter plus an update document built with operators such as $set, $inc or $unset. update_one() modifies the first match, update_many() modifies all matches, and replace_one() swaps the whole document (apart from _id). All three return an UpdateResult exposing matched_count and modified_count. Pass upsert=True to insert when nothing matches.

from pymongo import ReturnDocument

# Update the first match
res = col.update_one({"name": "John"}, {"$set": {"name": "Joseph"}})
print(res.matched_count, res.modified_count)

# Update every match, and bump a numeric field
col.update_many({"salary": 100}, {"$inc": {"salary": 10}})

# Upsert: update if present, otherwise insert
col.update_one({"name": "Alice"}, {"$set": {"salary": 90}}, upsert=True)

# Replace the whole document (keeps the same _id)
col.replace_one({"name": "Joseph"}, {"name": "George", "salary": 130})

# Atomically update and return the modified document
col.find_one_and_update(
    {"name": "George"},
    {"$inc": {"salary": 5}},
    return_document=ReturnDocument.AFTER,
)

Delete — delete_one() and delete_many()

delete_one() removes the first matching document and delete_many() removes all matches; both return a DeleteResult with a deleted_count. Calling delete_many({}) empties the collection, so use it carefully.

res = col.delete_one({"name": "John"})
print(res.deleted_count)

col.delete_many({"salary": {"$lt": 100}})

PyMongo 3.x methods removed in 4.x

If you are upgrading older code, these legacy helpers were removed in PyMongo 4.0. Swap them for the supported equivalents below before upgrading, or your code will raise AttributeError.

Removed (3.x)	Use instead (4.x)
insert()	insert_one() / insert_many()
update()	update_one() / update_many() / replace_one()
remove()	delete_one() / delete_many()
save()	insert_one() or replace_one(..., upsert=True)
find_and_modify()	find_one_and_update() / find_one_and_replace() / find_one_and_delete()
count()	count_documents() / estimated_document_count()
ensure_index()	create_index()

Indexes

Indexes keep queries fast and can enforce uniqueness. Create them with create_index() (the old ensure_index() is gone). Index creation is idempotent — re-running it is a no-op when the index already exists.

from pymongo import ASCENDING, DESCENDING

# Single-field unique index
col.create_index("name", unique=True)

# Compound index: salary descending, then name ascending
col.create_index([("salary", DESCENDING), ("name", ASCENDING)])

# Inspect existing indexes
print(col.index_information())

Aggregation pipeline

For grouping, filtering and transformations, use aggregate() with a pipeline of stages ($match, $group, $sort, $project, and more). The example below averages salary per name:

pipeline = [
    {"$match": {"salary": {"$gte": 100}}},
    {"$group": {"_id": "$name", "avg_salary": {"$avg": "$salary"}}},
    {"$sort": {"avg_salary": -1}},
]

for row in col.aggregate(pipeline):
    print(row["_id"], row["avg_salary"])

Bulk writes

When you have many mixed operations, batch them with bulk_write() to cut round-trips to the server. Pass a list of operation models; set ordered=False to let the server apply them in parallel and continue past individual errors.

from pymongo import InsertOne, UpdateOne, DeleteOne

ops = [
    InsertOne({"name": "Nina", "salary": 110}),
    UpdateOne({"name": "Steve"}, {"$set": {"salary": 125}}),
    DeleteOne({"name": "David"}),
]

result = col.bulk_write(ops, ordered=False)
print(result.inserted_count, result.modified_count, result.deleted_count)

Working with ObjectId

Every document gets a unique _id. When an id arrives from a client as a string, convert it back to an ObjectId before querying — and guard against malformed values:

from bson import ObjectId
from bson.errors import InvalidId

doc_id = "57611a711aa303032ad5ba9b"
col.find_one({"_id": ObjectId(doc_id)})

try:
    oid = ObjectId(doc_id)
except InvalidId:
    oid = None   # not a valid 24-char hex id

Async MongoDB with PyMongo

Need non-blocking I/O for an async web framework such as FastAPI? PyMongo ships a native async API (added in PyMongo 4.9) via AsyncMongoClient. It mirrors the synchronous API — you simply await each operation. The separate Motor driver is being folded into PyMongo's async support, so new projects should prefer the built-in async client.

import asyncio
from pymongo import AsyncMongoClient

async def main():
    client = AsyncMongoClient("mongodb://localhost:27017")
    col = client["test"]["person"]

    await col.insert_one({"name": "John", "salary": 100})
    doc = await col.find_one({"name": "John"})
    print(doc)

    async for d in col.find({"salary": {"$gte": 100}}):
        print(d["name"])

    await client.close()

asyncio.run(main())

Connection best practices

• Reuse one MongoClient for the whole process — it is thread-safe and pools connections internally. Don't create a new client per request.
• Keep credentials in environment variables (or a secrets manager), never in source control.
• Close the client on shutdown with client.close(), or use it as a context manager.
• Add type hints for clarity — MongoClient, Database and Collection are generic over the document type, e.g. Collection[dict].
• Tune the pool with maxPoolSize / minPoolSize on the client when you have specific concurrency needs.

Building with MongoDB and Python

At MicroPyramid we have shipped Python and MongoDB systems for startups and enterprises for 12+ years across 50+ projects — from data pipelines and analytics backends to high-throughput APIs. If you need a hand designing schemas, tuning indexes, or building a Python/MongoDB backend, our Python development team can help.

Frequently Asked Questions

Is PyMongo the official MongoDB driver for Python?

Yes. PyMongo is developed and maintained by MongoDB itself. It is the recommended way to talk to MongoDB from synchronous Python, and since version 4.9 it also includes a native asynchronous client.

How do I connect PyMongo to MongoDB Atlas?

Install the SRV extra with pip install "pymongo[srv]" and pass your Atlas mongodb+srv:// connection string to MongoClient. Store that string in an environment variable rather than hard-coding the username and password in your code.

Why does insert() or update() raise AttributeError in PyMongo 4?

The legacy insert(), update(), remove(), save(), find_and_modify() and count() methods were removed in PyMongo 4.0. Replace them with insert_one()/insert_many(), update_one()/update_many(), delete_one()/delete_many(), find_one_and_update() and count_documents().

How do I count documents in PyMongo 4.x?

Use col.count_documents(filter) for an exact count, or col.estimated_document_count() for a fast metadata-based estimate of the whole collection. The old count() method no longer exists.

Should I create a new MongoClient for every request?

No. Create one MongoClient and reuse it for the life of the process. It is thread-safe and manages its own connection pool, so a client per request wastes connections and hurts performance.

Does PyMongo support async/await?

Yes. PyMongo 4.9 introduced a native AsyncMongoClient whose API mirrors the synchronous one. The standalone Motor driver is being merged into PyMongo, so new async projects should use the built-in async client.