MongoDB: When Documents Are the Right Fit — Databases (Production)

Today, Postgres JSONB covers most of the jobs that MongoDB used to be chosen for. Around 2014, "schemaless" was MongoDB's main selling point. Since then, many teams have found, through real problems, that data without a schema has no rules, and so nothing stops bad data from getting in. Use Mongo when your data really is hierarchical and mostly read, when your collections need to grow across more machines than one Postgres server can comfortably handle, or when you have inherited a Mongo deployment and rewriting it is not an option.

Use Postgres + JSONB (db-postgres) for everything else. But when Mongo is the right tool, this lesson shows you how to use it well.

Examples assume a running local MongoDB. Docker is the easiest way to get one. See devops-docker. Pip-install the driver as shown. Expected output is in comments.

1. The Document Model

Mongo stores BSON documents: JSON plus a few extra types (binary ObjectId, native datetime, Decimal128, binary blobs). A collection is an unordered group of documents, and a database is a group of collections.

json

{
  "_id": ObjectId("6553f1..."),
  "title": "Postgres vs Mongo in 2026",
  "author_id": ObjectId("6553f0..."),
  "tags": ["databases", "python"],
  "comments": [
    {"author": "Alice", "text": "Good post", "at": ISODate("2026-04-12T...")},
    {"author": "Alex",  "text": "Counterpoint", "at": ISODate("2026-04-13T...")}
  ],
  "stats": {"views": 4210, "shares": 33}
}

Two design decisions to know from the start:

_id is mandatory and unique per collection. Mongo generates an ObjectId if you don't supply one. If you have a natural key, you can use any unique value instead (strings, integers, UUIDs).
Embedded vs referenced: comments are stored inside the post here. That's "embedding". author_id points to a document in another collection. That's "referencing". Choosing between the two is the main decision in Mongo data modelling (Section 9).

2. Connecting with PyMongo

bash

pip install pymongo

python

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["myapp"]
users = db["users"]

users.insert_one({"name": "Alice", "email": "alice@example.com", "active": True})
print(users.find_one({"email": "alice@example.com"}))
# {'_id': ObjectId('...'), 'name': 'Alice', 'email': 'alice@example.com', 'active': True}

MongoClient already manages a connection pool, so share one instance per process. If you create a new client for each request, you lose the benefit of the pool. The connection string accepts every Mongo option. Production deployments use a SRV record and TLS:

python

mongodb+srv://user:pass@cluster.example.net/myapp?retryWrites=true&w=majority

retryWrites=true and w=majority are the settings you want for any production cluster. A write that fails because of a network error is retried once, and a write only counts as done when a majority of replicas have it.

3. CRUD

The four basic operations: insert, find, update and delete.

Insert:

python

users.insert_one({"name": "Alice", "email": "s@example.com"})
users.insert_many([
    {"name": "Alex", "email": "alex@example.com"},
    {"name": "Kira", "email": "kira@example.com"},
])

+ setup added so this can run · defines users

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

users = _AutoMock('users')

Find:

python

users.find_one({"email": "s@example.com"})
list(users.find({"active": True}).limit(10))
list(users.find({"active": True}, {"name": 1, "email": 1, "_id": 0}))   # projection: return only name and email

+ setup added so this can run · defines users

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

users = _AutoMock('users')

find() returns a cursor. Loop over it, and don't call list() on it when the result is big. Pass a projection (the second argument) to choose which fields come back, like SELECT col, col in SQL.

Update:

python

users.update_one(
    {"_id": user_id},
    {"$set": {"name": "Alice T.", "active": True}},
)

users.update_many(
    {"active": False},
    {"$set": {"archived": True}},
)

+ setup added so this can run · defines users, user_id

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

users = _AutoMock('users')
user_id = _AutoMock('user_id')

Delete:

python

users.delete_one({"_id": user_id})
users.delete_many({"archived": True})

+ setup added so this can run · defines users, user_id

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

users = _AutoMock('users')
user_id = _AutoMock('user_id')

update_one and delete_one modify at most one document. The _many variants act on every match. If your filter is {}, that means every document in the collection. It is easy to delete everything by accident, so check every multi-document operation before it runs.

4. Query Operators

Where SQL has WHERE col > 10 AND col < 20, Mongo has nested dollar-prefixed operators:

python

users.find({"age": {"$gte": 18, "$lt": 65}})
users.find({"country": {"$in": ["UK", "IE"]}})
users.find({"email": {"$regex": "@example\\.com$"}})
users.find({"deleted_at": {"$exists": False}})
users.find({"$and": [{"active": True}, {"verified": True}]})
users.find({"$or": [{"role": "admin"}, {"role": "owner"}]})

+ setup added so this can run · defines users

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

users = _AutoMock('users')

The common ones to memorise:

Operator	Meaning
`$eq`, `$ne`	Equal / not equal
`$gt`, `$gte`, `$lt`, `$lte`	Comparisons
`$in`, `$nin`	In / not in a list
`$exists`	Field is present (or not)
`$regex`	Regex match
`$and`, `$or`, `$not`, `$nor`	Logical combination

5. Update Operators

$set is only one of them. These atomic update operators are why Mongo is good at counters and append-only data:

Operator	Meaning
`$set`	Assign field
`$unset`	Remove field
`$inc`	Increment numerically
`$push`	Append to an array
`$pull`	Remove matching items from an array
`$addToSet`	Append to an array if not already present
`$pop`	Remove first or last element

python

posts.update_one(
    {"_id": post_id},
    {
        "$inc": {"stats.views": 1},
        "$push": {"comments": {"author": "Alice", "text": "Hello", "at": datetime.utcnow()}},
    },
)

+ setup added so this can run · defines posts, post_id, datetime

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

posts = _AutoMock('posts')
post_id = _AutoMock('post_id')
datetime = _AutoMock('datetime')

Both updates above are atomic at the document level. Mongo guarantees that no other client sees that document with only part of the update applied.

6. Aggregation Pipeline

The aggregation pipeline is how you analyse data in Mongo. It is a list of stages, run in order, and each stage transforms the documents it receives.

python

from pymongo import DESCENDING

result = users.aggregate([
    {"$match": {"active": True}},
    {"$group": {"_id": "$country", "count": {"$sum": 1}, "avg_age": {"$avg": "$age"}}},
    {"$sort": {"count": DESCENDING}},
    {"$limit": 10},
])

for row in result:
    print(row)
# {'_id': 'UK', 'count': 1240, 'avg_age': 32.4}
# {'_id': 'US', 'count':  980, 'avg_age': 35.1}
# ...

+ setup added so this can run · defines users

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

users = _AutoMock('users')

Key stages:

Stage	Equivalent
`$match`	`WHERE`
`$group`	`GROUP BY`
`$project`	`SELECT col1, col2`
`$sort`	`ORDER BY`
`$limit` / `$skip`	`LIMIT` / `OFFSET`
`$lookup`	`JOIN` (slow vs SQL, see Section 9)
`$unwind`	Explode an array field into one doc per element

In apps that use only Mongo, the pipeline is how you build reports. For heavy analytics, you would usually copy the data (ETL) to a dedicated OLAP database instead.

7. Indexes

The default _id index is the only one Mongo creates for you. Any other query on a collection of real size needs an index. Without one, Mongo scans the whole collection.

python

from pymongo import ASCENDING, DESCENDING

users.create_index([("email", ASCENDING)], unique=True)
posts.create_index([("author_id", ASCENDING), ("created", DESCENDING)])
posts.create_index([("title", "text")])               # text search index

+ setup added so this can run · defines users, posts

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

users = _AutoMock('users')
posts = _AutoMock('posts')

Compound index order matters. A (author_id, created) index supports queries on author_id alone and on author_id + created, but not on created alone. This is the same prefix rule as for SQL composite indexes.

To find a missing index, run cursor.explain() and look for COLLSCAN in the plan. If you see it on a query that runs often, add the index.

unique=True enforces uniqueness in the database itself. That is the only place where uniqueness can be guaranteed.

8. Async with Motor

pymongo is sync. For asyncio apps, use Motor (the official async driver):

bash

pip install motor

python

from motor.motor_asyncio import AsyncIOMotorClient
import asyncio

async def main():
    client = AsyncIOMotorClient("mongodb://localhost:27017/")
    users = client.myapp.users

    await users.insert_one({"name": "Alice", "active": True})

    async for user in users.find({"active": True}):
        print(user)

    client.close()

asyncio.run(main())

+ setup added so this can run · defines user

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

user = _AutoMock('user')

Motor has almost the same API as PyMongo, call for call. Every operation is awaitable and find() returns an async iterator.

If you'd rather use Pydantic models on top, Beanie (built on Motor) and odmantic are the current ODM choices.

9. Embed or Reference

This modelling choice decides, more than anything else, whether your Mongo app stays fast.

Embed when:

The child data only matters in the context of the parent (comments on a post, line items on an order).

One-to-few: under a few hundred children.

The parent is the natural read unit ("show me a post" naturally wants its comments).

Total document stays well under the 16 MB BSON limit.

Reference when:

The child is independently queryable (users → posts, where you query "all posts by X").

One-to-many or many-to-many at any scale.

The child is shared across parents.

python

# Embedded: one read gets everything needed to show the post page
{
  "_id": post_id,
  "title": "...",
  "comments": [{"author": "Alice", "text": "..."}]
}

# Referenced: the author is stored in users, and posts link to it by author_id
{"_id": post_id, "title": "...", "author_id": author_obj_id}
{"_id": author_obj_id, "name": "Alice", "email": "..."}

+ setup added so this can run · defines post_id, author_obj_id

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

post_id = _AutoMock('post_id')
author_obj_id = _AutoMock('author_obj_id')

$lookup joins them in the aggregation pipeline, but it is much slower than a SQL JOIN, because Mongo was not designed for joins. If you need $lookup all the time, that is a warning sign: your data is relational, and Postgres is the right tool.

10. Transactions

Mongo transactions exist (since 4.0) and span multiple documents and collections. They require a replica set (a single standalone Mongo server doesn't support them), and they make operations slower.

python

with client.start_session() as session:
    with session.start_transaction():
        accounts.update_one({"_id": from_id}, {"$inc": {"balance": -amount}}, session=session)
        accounts.update_one({"_id": to_id},   {"$inc": {"balance":  amount}}, session=session)
        # commits if the block ends normally, aborts on an exception

+ setup added so this can run · defines client, accounts, from_id, to_id, amount

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

client = _AutoMock('client')
accounts = _AutoMock('accounts')
from_id = _AutoMock('from_id')
to_id = _AutoMock('to_id')
amount = 1

Use as few transactions as you can. Mongo is designed around one atomic document at a time. If you often need several documents to change atomically, the data is probably relational, and again Postgres is the right tool.

11. When Mongo Fails You

Know where Mongo stops being a good fit.

Relational queries across collections: $lookup works, but slowly, and the aggregation pipelines quickly become hard to read.
Strong consistency across multiple documents: possible with transactions, but at a cost. In Postgres this is normal and cheap.
Mature analytics tools: every BI tool supports SQL directly. Many support Mongo only through translation layers that lose features.
Enforcing data shapes: Postgres NOT NULL / CHECK constraints and foreign keys stop whole classes of bugs in the database. Mongo has schema validation (based on JSON Schema), but it is optional and checks less.
Little memory: Mongo needs a lot of RAM. The data you use most often should fit in memory, or performance suffers.

If you are three months into a project and keep needing features that Mongo lacks or runs slowly, moving to Postgres is a lot of work, but it is usually the right decision.

Common Mistakes

1. No indexes on common queries. Without an index, every query scans the whole collection. On a collection of 10 million documents, that can take hundreds of milliseconds per query, and under heavy traffic the service can stop responding. Use explain() and create_index() from the first day.

2. Storing dates as strings. "2026-05-14" sorts lexically, which works for ISO 8601 strings, until someone writes "14/05/2026". Use real datetime objects, because Mongo stores them as BSON dates and indexes them properly.

3. Over-embedding. A document that grows without limit (every event in a user's life embedded in the user document) reaches the 16 MB cap and makes every read very slow. Keep embedded arrays small and limited in size. Comments on a post are fine to embed, but every event forever is not.

4. Modelling like a relational DB with refs everywhere. Every document points to three others and every operation is three round trips. Mongo is weak at this, and it is a sign that you should be using Postgres. Embed where Mongo wants you to embed.

5. Using Mongo when Postgres+JSONB would do. Almost everyone can say "our data is hierarchical". JSONB in Postgres gives you hierarchical storage plus joins, transactions, constraints, and mature tooling.

6. One MongoClient per request. MongoClient is a connection pool. Build one per process at startup, and share it.

7. Not setting _id deliberately. Letting Mongo generate ObjectIds is fine. But if your natural key is the email, use the email as _id and skip the secondary unique index.

🎯 Your Turn: Blog Schema with Embedded Comments and Referenced Authors

Model a tiny blog: users and posts. Embed comments inside posts (small, read-with-post). Reference author_id to link posts to users. Implement four functions.

Skeleton:

python

from datetime import datetime
from pymongo import MongoClient, ASCENDING
from bson import ObjectId

client = MongoClient("mongodb://localhost:27017/")
db = client["blogdemo"]
users = db["users"]
posts = db["posts"]

users.create_index([("email", ASCENDING)], unique=True)
posts.create_index([("author_id", ASCENDING)])

def create_user(name: str, email: str) -> ObjectId:
    # TODO 1
    ...

def create_post(author_id: ObjectId, title: str, body: str) -> ObjectId:
    # TODO 2: insert with an empty comments array and return the new id
    ...

def add_comment(post_id: ObjectId, author_name: str, text: str) -> None:
    # TODO 3: $push a comment onto the embedded comments array
    ...

def posts_by_author_name(name: str) -> list[dict]:
    # TODO 4: build a pipeline that matches the user by name, does a $lookup on posts,
    # and returns the post docs with the author name attached
    ...

Hint 1: $push and the comments array

posts.update_one({"_id": post_id}, {"$push": {"comments": {...}}}). Store the comment as a small embedded doc with author, text, and an at datetime.

Hint 2: $lookup as a JOIN

Start the pipeline on the users collection ($match by name), then $lookup from posts using localField: "_id" and foreignField: "author_id". The matched posts come back nested under a new field. Use $unwind to flatten and $project to shape the output.

Show full solution

python

from datetime import datetime
from pymongo import MongoClient, ASCENDING
from bson import ObjectId

client = MongoClient("mongodb://localhost:27017/")
db = client["blogdemo"]
users = db["users"]
posts = db["posts"]

users.create_index([("email", ASCENDING)], unique=True)
posts.create_index([("author_id", ASCENDING)])


def create_user(name: str, email: str) -> ObjectId:
    res = users.insert_one({"name": name, "email": email, "created": datetime.utcnow()})
    return res.inserted_id


def create_post(author_id: ObjectId, title: str, body: str) -> ObjectId:
    res = posts.insert_one({
        "author_id": author_id,
        "title": title,
        "body": body,
        "comments": [],
        "created": datetime.utcnow(),
    })
    return res.inserted_id


def add_comment(post_id: ObjectId, author_name: str, text: str) -> None:
    posts.update_one(
        {"_id": post_id},
        {"$push": {"comments": {"author": author_name, "text": text, "at": datetime.utcnow()}}},
    )


def posts_by_author_name(name: str) -> list[dict]:
    pipeline = [
        {"$match": {"name": name}},
        {"$lookup": {
            "from": "posts",
            "localField": "_id",
            "foreignField": "author_id",
            "as": "posts",
        }},
        {"$unwind": "$posts"},
        {"$project": {
            "_id": "$posts._id",
            "title": "$posts.title",
            "body": "$posts.body",
            "author": "$name",
            "comments": "$posts.comments",
            "created": "$posts.created",
        }},
    ]
    return list(users.aggregate(pipeline))


# Demo
surya_id = create_user("Alice", "alice@example.com")
post_id  = create_post(surya_id, "First post", "Hello, Mongo")
add_comment(post_id, "Alex", "Nice one")
add_comment(post_id, "Kira", "Welcome")

for p in posts_by_author_name("Alice"):
    print(p["title"], "by", p["author"], "with", len(p["comments"]), "comments")
# First post by Alice with 2 comments

What the solution gets right:

Indexes from the start: email unique on users, author_id on posts. Both queries we run hit indexes.
Embedded comments: a small, limited array per post, so one read fetches the post with its comments.
Referenced author: authors are independent entities you might list, edit, or query on their own.
$push for comments: atomic at the document level, so two comments added at the same time can't overwrite each other.
Aggregation $lookup for the join: works fine here because the lookup is one-to-many and the join key is indexed.
Datetimes as real datetime objects, not strings.

For a high-traffic version, denormalise the author name onto the post itself ({"author_id": ..., "author_name": "Alice"}) and update both places when the user changes their name. Writes become more complex, but reads become faster. Most Mongo modelling is about choosing trade-offs like this one.

What You Learned

Documents and BSON: JSON plus ObjectId, datetime, Decimal128, binary.
One MongoClient per process, because it already has a connection pool.
CRUD with _one vs _many: the _many variants can change many documents at once, so use them with care.
Query operators ($gt, $in, $exists, $regex, $and, $or) written as nested dicts.
Update operators ($set, $inc, $push, $pull, $addToSet), atomic at the document level.
Aggregation pipelines for grouping, sorting, projection, and (carefully) joins via $lookup.
Indexes are mandatory, and compound-index order follows the prefix rule.
Async via Motor, with ODMs Beanie and odmantic on top.
Embed for read-with-parent, reference for independent or shared data.
Transactions exist, but needing them often is a sign of relational data. Use them rarely.
Check the fit: Postgres + JSONB now does most of the jobs Mongo was once chosen for.

Next: ORM Patterns That Work in Production: N+1, eager loading, bulk ops, migrations, and when to drop to raw SQL.