MongoDB #

MongoDB is a document-based NoSQL database that stores data in BSON (Binary JSON) format — not rows and columns like relational databases, but flexible documents that can contain nested structures. This makes it ideal for data with frequently changing schemas, complex documents, and applications that need horizontal scaling. Python interacts with MongoDB through the PyMongo library — the official driver providing a complete API from CRUD operations and aggregation pipelines to multi-document transactions. Understanding how MongoDB’s query operators work and how the aggregation pipeline operates is key to using MongoDB optimally.

The most fundamental difference between relational (SQL) databases and document-based databases (MongoDB/BSON) lies in how data is organized. While SQL stores data as flat rows and needs relationships (JOINs) between separate tables, MongoDB stores data in a single complete BSON document supporting nested objects, as shown in the diagram below:

flowchart TD
    subgraph SQL ["Relational Database (SQL - Separate Tables)"]
        Row1["Users Table: id, name, email"]
        Row2["Addresses Table: id, user_id, city, street"]
        Row1 -->|"JOIN (Relationship)"| Row2
    end

    subgraph MongoDB ["Document Database (MongoDB - Nested BSON)"]
        Doc["BSON Document (One Entity)<br>{<br>&nbsp;&nbsp;id: 1,<br>&nbsp;&nbsp;nama: 'Budi',<br>&nbsp;&nbsp;alamat: {<br>&nbsp;&nbsp;&nbsp;&nbsp;kota: 'Jakarta',<br>&nbsp;&nbsp;&nbsp;&nbsp;jalan: 'Sudirman'<br>&nbsp;&nbsp;}<br>}"]
    end

Installation #

pip install pymongo

For connecting to MongoDB Atlas (cloud) or when TLS is required, add the optional dependency:

pip install "pymongo[srv]"   # for mongodb+srv:// connection strings

Creating a Connection #

MongoClient manages the connection pool automatically — a single instance is enough for the whole application and is thread-safe.

from pymongo import MongoClient
import os

# ANTI-PATTERN: hardcoding the connection in code
client = MongoClient("localhost", 27017)  # ✗ -- not flexible for deployment

# CORRECT: read from an environment variable
MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017")
client = MongoClient(
    MONGO_URI,
    serverSelectionTimeoutMS=5000,   # timeout if the server can't be reached
    connectTimeoutMS=10000,
    maxPoolSize=50                   # connection pool limit
)

# Connecting to MongoDB Atlas
# MONGO_URI = "mongodb+srv://user:***@cluster.mongodb.net/myapp?retryWrites=true"
client_atlas = MongoClient(os.getenv("MONGO_URI"))

# Test the connection
try:
    client.admin.command("ping")
    print("MongoDB connection successful.")
except Exception as e:
    print(f"Connection failed: {e}")

# Access the database and collections
db         = client["myapp"]          # or client.myapp
pengguna   = db["pengguna"]           # or db.pengguna
produk     = db["produk"]
orders     = db["orders"]

MongoClient is thread-safe and manages the connection pool internally. Create a single instance at the application level and share it across all code — don’t create a new instance per request or per function.

Document Structure #

Unlike relational tables, MongoDB documents can contain diverse data types and nested structures. Understanding this structure is important before writing queries.

# Example user document
dokumen_pengguna = {
    # _id is automatically created as an ObjectId if not included
    "nama":      "Budi Santoso",
    "email":     "[email protected]",
    "usia":      28,
    "aktif":     True,
    "tag":       ["python", "backend"],       # array
    "alamat": {                               # embedded document
        "kota":     "Bandung",
        "provinsi": "Jawa Barat",
        "kode_pos": "40115"
    },
    "riwayat_login": [                        # array of embedded documents
        {"waktu": "2024-03-15T10:00:00", "ip": "192.168.1.1"},
        {"waktu": "2024-03-16T08:30:00", "ip": "192.168.1.2"},
    ]
}

# Example product document
dokumen_produk = {
    "nama":     "Laptop Gaming ASUS ROG",
    "harga":    15000000,
    "stok":     5,
    "kategori": "Elektronik",
    "spesifikasi": {
        "ram":       "16GB",
        "storage":   "512GB SSD",
        "processor": "Intel i7"
    },
    "tag": ["laptop", "gaming", "asus"]
}

Inserting Documents #

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

# Insert one document
def tambah_pengguna(db, nama: str, email: str, usia: int) -> str:
    doc = {
        "nama":       nama,
        "email":      email,
        "usia":       usia,
        "aktif":      True,
        "tag":        [],
        "dibuat_pada": datetime.now(timezone.utc)
    }
    hasil = db["pengguna"].insert_one(doc)
    return str(hasil.inserted_id)

id_baru = tambah_pengguna(db, "Budi Santoso", "[email protected]", 28)
print(f"New ID: {id_baru}")

# Insert many documents at once
def tambah_banyak_pengguna(db, daftar: list[dict]) -> list[str]:
    for doc in daftar:
        doc["dibuat_pada"] = datetime.now(timezone.utc)
        doc.setdefault("aktif", True)
        doc.setdefault("tag", [])
    
    hasil = db["pengguna"].insert_many(daftar)
    return [str(oid) for oid in hasil.inserted_ids]

data_baru = [
    {"nama": "Sari Dewi",     "email": "[email protected]",  "usia": 25},
    {"nama": "Andi Prasetyo", "email": "[email protected]",  "usia": 32},
    {"nama": "Rina Marlina",  "email": "[email protected]",  "usia": 29},
]
ids = tambah_banyak_pengguna(db, data_baru)
print(f"{len(ids)} documents added.")

Read — Query Operators #

MongoDB uses dict-based operators to filter documents. This is very different from an SQL WHERE clause.

Comparison Operators #

col = db["pengguna"]

# Equal (implicit)
col.find_one({"email": "[email protected]"})

# Explicit operators
col.find({"usia": {"$gt":  25}})          # usia > 25
col.find({"usia": {"$gte": 25}})          # usia >= 25
col.find({"usia": {"$lt":  30}})          # usia < 30
col.find({"usia": {"$lte": 30}})          # usia <= 30
col.find({"usia": {"$ne":  28}})          # usia != 28
col.find({"usia": {"$in":  [25, 28, 32]}})  # usia IN (25, 28, 32)
col.find({"usia": {"$nin": [25, 28]}})    # usia NOT IN (25, 28)

# Range
col.find({"usia": {"$gte": 20, "$lte": 30}})  # 20 <= usia <= 30

# Null and field existence
col.find({"foto": None})                       # foto == null
col.find({"foto": {"$exists": False}})         # the foto field doesn't exist at all
col.find({"foto": {"$exists": True}})          # the foto field exists (even if null)

Logical Operators #

# $and -- all conditions must hold (default when there are multiple keys)
col.find({"aktif": True, "usia": {"$gte": 25}})  # implicit AND

# $or -- at least one condition holds
col.find({"$or": [
    {"nama": {"$regex": "budi", "$options": "i"}},
    {"email": {"$regex": "budi", "$options": "i"}}
]})

# $nor -- none of the conditions hold
col.find({"$nor": [{"aktif": False}, {"usia": {"$lt": 18}}]})

# $not
col.find({"usia": {"$not": {"$lt": 18}}})  # usia not less than 18

Querying Nested Fields and Arrays #

# Nested fields (embedded documents) -- use dot notation
col.find({"alamat.kota": "Bandung"})
col.find({"alamat.kode_pos": {"$regex": "^40"}})

# Arrays -- contains
col.find({"tag": "python"})                    # tag contains "python"
col.find({"tag": {"$in": ["python", "go"]}})   # tag contains python OR go
col.find({"tag": {"$all": ["python", "backend"]}})  # tag contains both

# Array size
col.find({"tag": {"$size": 0}})               # empty tag

Projection — Selecting Specific Fields #

# ANTI-PATTERN: fetching whole documents when only a few fields are needed
semua = list(col.find({"aktif": True}))        # ✗ -- fetches every field, wasteful bandwidth

# CORRECT: use a projection to limit the returned fields
# 1 = include, 0 = exclude (can't mix except for _id)
semua = list(col.find(
    {"aktif": True},
    {"nama": 1, "email": 1, "usia": 1, "_id": 0}  # ✓ -- only nama, email, usia
))

# Exclude large fields
semua = list(col.find(
    {},
    {"riwayat_login": 0, "bio": 0}   # fetch everything except these two fields
))

Sorting and Pagination #

import pymongo

# Sorting
col.find().sort("nama", pymongo.ASCENDING)    # A-Z
col.find().sort("nama", pymongo.DESCENDING)   # Z-A
col.find().sort([                             # multi-column
    ("usia", pymongo.DESCENDING),
    ("nama", pymongo.ASCENDING)
])

# Pagination
halaman     = 2
per_halaman = 10
offset      = (halaman - 1) * per_halaman

hasil = list(
    col.find({"aktif": True})
    .sort("dibuat_pada", pymongo.DESCENDING)
    .skip(offset)
    .limit(per_halaman)
)

# Count the total for pagination info
total = col.count_documents({"aktif": True})
print(f"Page {halaman}, showing {len(hasil)} of {total} users")

Updating Documents #

MongoDB provides various update operators that work on specific fields without overwriting the whole document.

col = db["pengguna"]

# ANTI-PATTERN: replacing the whole document
col.update_one(
    {"email": "[email protected]"},
    {"nama": "Budi Wijaya"}         # ✗ -- overwrites the whole document, other fields lost!
)

# CORRECT: use $set to update specific fields
col.update_one(
    {"email": "[email protected]"},
    {"$set": {"nama": "Budi Wijaya", "usia": 29}}   # ✓ -- only these fields change
)

# Common update operators
col.update_one(
    {"email": "[email protected]"},
    {
        "$set":       {"nama": "Budi Wijaya"},        # set a field's value
        "$inc":       {"usia": 1},                    # increment (+1)
        "$push":      {"tag": "devops"},              # add to an array
        "$addToSet":  {"tag": "backend"},             # add to an array if not present
        "$pull":      {"tag": "junior"},              # remove from an array
        "$unset":     {"foto": ""},                   # remove a field
        "$currentDate": {"diubah_pada": True}         # set to the current time
    }
)

# upsert -- update if it exists, insert if not
col.update_one(
    {"email": "[email protected]"},
    {"$set": {"nama": "Budi", "aktif": True}},
    upsert=True
)

# Update many documents at once
hasil = col.update_many(
    {"aktif": False, "usia": {"$lt": 18}},
    {"$set": {"label": "nonaktif-minor"}}
)
print(f"{hasil.modified_count} documents updated.")

# findOneAndUpdate -- fetch the old document before/after the update
from pymongo import ReturnDocument

dokumen_baru = col.find_one_and_update(
    {"email": "[email protected]"},
    {"$set": {"aktif": False}},
    return_document=ReturnDocument.AFTER   # BEFORE for the pre-update document
)

Deleting Documents #

col = db["pengguna"]

# Delete one document (the first one found)
hasil = col.delete_one({"email": "[email protected]"})
print(f"Deleted: {hasil.deleted_count} documents")

# Delete many documents
hasil = col.delete_many({"aktif": False})
print(f"Deleted: {hasil.deleted_count} inactive documents")

# findOneAndDelete -- delete and return the deleted document
dokumen_terhapus = col.find_one_and_delete({"email": "[email protected]"})
if dokumen_terhapus:
    print(f"Document deleted: {dokumen_terhapus['nama']}")

Aggregation Pipelines #

The aggregation pipeline is MongoDB’s most powerful feature — enabling data transformation through a series of stages executed sequentially.

col = db["orders"]

# Basic pipeline: filter → group → sort
pipeline = [
    # Stage 1: $match -- filter documents (like WHERE in SQL)
    {"$match": {"status": "selesai"}},

    # Stage 2: $group -- group and aggregate
    {"$group": {
        "_id":               "$pengguna_id",
        "jumlah_order":      {"$sum": 1},
        "total_belanja":     {"$sum": "$total"},
        "rata_belanja":      {"$avg": "$total"},
        "order_pertama":     {"$min": "$dibuat_pada"},
        "order_terakhir":    {"$max": "$dibuat_pada"},
    }},

    # Stage 3: $sort -- order the results
    {"$sort": {"total_belanja": -1}},

    # Stage 4: $limit -- cap the results
    {"$limit": 10}
]

hasil = list(col.aggregate(pipeline))
for r in hasil:
    print(f"User {r['_id']}: {r['jumlah_order']} orders, total Rp{r['total_belanja']:,.0f}")

Advanced Pipelines #

col_pengguna = db["pengguna"]

# $lookup -- JOIN to another collection
pipeline_lookup = [
    {"$match": {"aktif": True}},

    # JOIN orders to pengguna
    {"$lookup": {
        "from":         "orders",         # target collection
        "localField":   "_id",            # field in this collection
        "foreignField": "pengguna_id",    # field in the target collection
        "as":           "orders"          # name of the JOIN result field
    }},

    # $addFields -- add computed fields
    {"$addFields": {
        "jumlah_order": {"$size": "$orders"},
        "total_belanja": {"$sum": "$orders.total"}
    }},

    # $project -- select the returned fields
    {"$project": {
        "nama":          1,
        "email":         1,
        "jumlah_order":  1,
        "total_belanja": 1,
        "_id":           0
    }},

    {"$sort": {"total_belanja": -1}},
    {"$limit": 20}
]

hasil = list(col_pengguna.aggregate(pipeline_lookup))

# $unwind -- split an array into separate documents
pipeline_unwind = [
    {"$unwind": "$tag"},                    # split the tag array
    {"$group": {
        "_id":   "$tag",
        "count": {"$sum": 1}
    }},
    {"$sort": {"count": -1}},
    {"$limit": 10}
]

top_tag = list(db["pengguna"].aggregate(pipeline_unwind))
for t in top_tag:
    print(f"Tag '{t['_id']}': {t['count']} users")

Indexing #

Indexes are critical for query performance on large collections. Without an index, MongoDB performs a full collection scan for every query.

import pymongo

col = db["pengguna"]

# Single-field indexes
col.create_index("email", unique=True)               # unique index for email
col.create_index("aktif")                            # regular index
col.create_index([("dibuat_pada", pymongo.DESCENDING)])  # descending

# Compound index -- field order matters!
col.create_index([
    ("aktif", pymongo.ASCENDING),
    ("dibuat_pada", pymongo.DESCENDING)
])

# Text index -- for full-text search
col.create_index([("nama", pymongo.TEXT), ("bio", pymongo.TEXT)])

# Text queries with a text index
col.find({"$text": {"$search": "budi santoso"}})
col.find({"$text": {"$search": "\"budi santoso\""}})   # exact phrase

# TTL index -- documents automatically deleted after N seconds
db["sesi"].create_index(
    "dibuat_pada",
    expireAfterSeconds=3600    # delete documents 1 hour after dibuat_pada
)

# List all indexes in a collection
print(list(col.list_indexes()))

# Drop an index
col.drop_index("email_1")

Don’t create excessive indexes. Each index speeds up reads but slows down writes because MongoDB must update all indexes on insert/update/delete. Create indexes only for fields frequently used in query filters or sorts. Use explain() to analyze whether a query uses an index correctly.
# Analyze query execution
col.find({"aktif": True, "usia": {"$gte": 25}}).explain("executionStats")

Error Handling #

from pymongo.errors import (
    DuplicateKeyError,
    ConnectionFailure,
    OperationFailure,
    ServerSelectionTimeoutError
)

def tambah_pengguna_aman(db, nama: str, email: str) -> str | None:
    try:
        hasil = db["pengguna"].insert_one({
            "nama":  nama,
            "email": email,
            "aktif": True
        })
        return str(hasil.inserted_id)
    
    except DuplicateKeyError:
        print(f"Email '{email}' is already registered.")
        return None
    
    except OperationFailure as e:
        print(f"Operation failed: {e.details}")
        return None

def koneksi_aman(uri: str):
    try:
        client = MongoClient(uri, serverSelectionTimeoutMS=3000)
        client.admin.command("ping")
        return client
    except ServerSelectionTimeoutError:
        print("Can't reach the MongoDB server.")
        return None
    except ConnectionFailure as e:
        print(f"Connection failed: {e}")
        return None

When to Choose MongoDB vs a Relational Database #

Choose MongoDB when:
  ✓ The data schema changes often or is inconsistent between documents
  ✓ Data is hierarchical and often read as a single unit (embedded)
  ✓ You need easy horizontal scaling (sharding)
  ✓ Very large data volumes with high write-throughput needs
  ✓ Examples: e-commerce product catalogs, CMS content, application logs, user profiles

Choose a relational database (PostgreSQL, MySQL) when:
  ✓ Data is highly relational and ACID consistency is critical (financial transactions)
  ✓ The data schema is stable and well-defined
  ✓ You need complex JOIN queries across many entities
  ✓ The team is already familiar with SQL and its ecosystem

Summary #

One MongoClient for the whole application — the client is thread-safe and manages the connection pool; don’t create a new instance per request.
find_one() vs find() — find_one() returns a dict or None; find() returns a Cursor that must be iterated or converted with list().
The $set operator is mandatory for updates — without $set, the entire document is overwritten and all other fields are lost.
Projection for efficiency — always limit returned fields with a projection when you don’t need whole documents, especially for large fields like arrays or embedded documents.
Dot notation for nested fields — use "alamat.kota" to query or update fields inside an embedded document.
$addToSet vs $push — use $addToSet to add to an array without duplicates; $push always appends even if it already exists.
upsert=True — use it for idempotent insert-or-update operations without checking existence first.
Indexes for frequently filtered and sorted fields — create compound indexes with field order matching the most common query order.
Aggregation pipelines — use them for complex data transformations; putting $match first in the pipeline is crucial for using indexes and reducing processed documents.
TTL indexes — leverage them for data with an expiry, like sessions, tokens, or temporary logs.

← Previous: Django ORM Next: Elasticsearch →