MongoDB from Basics to Production · MongoDB in Production
Application Code with the Python Driver
Use PyMongo from an application: one client per process, write concerns, unique indexes with duplicate-key handling, bulk writes and timeouts.
What you'll build
The data layer of a sign-up service in Python: a single MongoClient with timeouts, a unique index that turns duplicate sign-ups into a clear error, an idempotent upsert, and a bulk write that imports many records in one round trip.
The idea
- Create one
MongoClientper process and reuse it; it manages a connection pool and server discovery. - Set timeouts (
serverSelectionTimeoutMS,timeoutMS) so requests fail fast when the database is unreachable. - Enforce uniqueness with a unique index and handle
DuplicateKeyError; don't check-then-insert. bulk_writesends many operations in one request;ordered=Falsecontinues past individual errors.- Retryable writes (on by default) retry once on network errors and failovers.
The code
import os
from pymongo import MongoClient, ASCENDING, UpdateOne, InsertOne
from pymongo.errors import BulkWriteError, DuplicateKeyError
client = MongoClient(os.environ["MONGO_URL"], serverSelectionTimeoutMS=3000, timeoutMS=5000, appname="signup-service")
db = client["signup_example"]
users = db.users
users.create_index([("email", ASCENDING)], unique=True)
def sign_up(email: str, name: str) -> str:
try:
users.insert_one({"email": email.lower(), "name": name})
return "created"
except DuplicateKeyError:
return "email already registered"
print(sign_up("[email protected]", "Ana"))
print(sign_up("[email protected]", "Ana again"))
# Idempotent: running it twice leaves one document with the latest values
for _ in range(2):
users.update_one({"email": "[email protected]"}, {"$set": {"name": "Ben"}, "$setOnInsert": {"plan": "free"}}, upsert=True)
print("ben documents:", users.count_documents({"email": "[email protected]"}))
ops = [InsertOne({"email": f"user{i}@example.com", "name": f"User {i}"}) for i in range(5)]
ops.append(InsertOne({"email": "[email protected]", "name": "duplicate"}))
try:
users.bulk_write(ops, ordered=False)
except BulkWriteError as e:
print("bulk inserted:", e.details["nInserted"], "| errors:", [err["code"] for err in e.details["writeErrors"]])
print("total users:", users.count_documents({}))
client.drop_database("signup_example")
Output
created email already registered ben documents: 1 bulk inserted: 5 | errors: [11000] total users: 7
Run this on your computer — see “Get Started” for setup.
Common mistakes
- Creating a new
MongoClientper request, exhausting connections and adding latency. - Check-then-insert for uniqueness, which races; use a unique index.
- No timeouts, so requests hang for 30 seconds when a node is unreachable.
- Ordered bulk writes that stop at the first error when the rest should still be imported.
Key points
- One client per process, with timeouts.
- Unique indexes plus DuplicateKeyError handling enforce uniqueness.
- bulk_write for many operations; upserts for idempotency.
Further reading: PyMongo: get started · Unique indexes · Retryable writes
Practice questions
Why should an application create one MongoClient and reuse it?
Answer
The client maintains a connection pool and monitors the cluster; creating one per request wastes connections and adds latency.
How do you guarantee unique emails?
Answer
Create a unique index on email and handle DuplicateKeyError on insert; check-then-insert races under concurrency.
What does ordered=False change in bulk_write?
Answer
The server continues with the remaining operations after an error and reports all errors at the end.
Open in the app to mark lessons complete, take notes and track your progress.