PyMongo là driver Python chính thức của MongoDB. Thông qua MongoClient, chương trình Python có thể kết nối tới MongoDB cục bộ hoặc Atlas, chọn database và collection, sau đó thực hiện đầy đủ các thao tác CRUD. Bài viết này xây dựng một ví dụ quản lý thư viện từ đầu đến cuối.

1. Cài đặt PyMongo

Nên tạo môi trường ảo để tách dependency của dự án, sau đó cài PyMongo bằng chính trình thông dịch sẽ chạy ứng dụng:

python -m venv .venv
.venv\Scripts\activate
python -m pip install --upgrade pip
python -m pip install pymongo

Trên macOS/Linux, lệnh kích hoạt là source .venv/bin/activate. Có thể kiểm tra bản cài bằng python -m pip show pymongo.

2. Kết nối Python với MongoDB

Với MongoDB đang chạy trên máy cục bộ ở cổng mặc định 27017, tạo client và gửi lệnh ping để xác nhận kết nối thật sự:

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure

uri = "mongodb://localhost:27017/"
client = MongoClient(uri, serverSelectionTimeoutMS=5000)

try:
    client.admin.command("ping")
    print("Đã kết nối MongoDB")
except ConnectionFailure as exc:
    print(f"Kết nối thất bại: {exc}")
    raise

MongoClient quản lý connection pool và nên được tái sử dụng trong vòng đời ứng dụng. Với Atlas, lấy URI mongodb+srv://... từ màn hình Connect và lưu trong biến môi trường; không hard-code username hoặc mật khẩu vào mã nguồn.

3. Tạo database, collection và chèn document

Việc truy cập client["library"] và db["library_details"] chỉ tạo đối tượng tham chiếu. Database và collection chỉ xuất hiện trên máy chủ sau thao tác ghi đầu tiên.

db = client["library"]
books = db["library_details"]

documents = [
    {
        "book_id": "1010",
        "book_name": "Python Programming",
        "book_author": "John M Zelle",
        "volume": 3,
    },
    {
        "book_id": "1019",
        "book_name": "Python for Data Analysis",
        "book_author": "Wes McKinney",
        "volume": 1,
    },
    {
        "book_id": "1009",
        "book_name": "Python Cookbook",
        "book_author": "David Beazley",
        "volume": 3,
    },
]

result = books.insert_many(documents)
print("ID đã chèn:", result.inserted_ids)

Nếu document không có _id, PyMongo tự tạo ObjectId duy nhất. Khi chạy lại đoạn chèn trên, chương trình sẽ tạo thêm document mới; trong ứng dụng thật nên dùng unique index hoặc upsert nếu cần tính lặp lại an toàn.

4. Đọc document bằng find_one và find

from pprint import pprint

first_book = books.find_one()
pprint(first_book)

for book in books.find({}):
    pprint(book)

find_one() trả về một dictionary hoặc None. find() trả về cursor, vì vậy nên lặp trực tiếp thay vì chuyển toàn bộ thành list khi collection lớn.

5. Lọc bằng điều kiện và toán tử

# Tìm sách của Wes McKinney
for book in books.find({"book_author": "Wes McKinney"}):
    pprint(book)

# volume lớn hơn 1
for book in books.find({"volume": {"$gt": 1}}):
    pprint(book)

# Kết hợp nhiều điều kiện: AND ngầm định
query = {
    "book_name": "Python Programming",
    "volume": {"$gt": 1},
}
for book in books.find(query):
    pprint(book)

Các trường nằm cùng cấp trong filter được kết hợp bằng AND ngầm định. Chỉ cần dùng $and tường minh khi phải kết hợp nhiều biểu thức trên cùng một trường hoặc khi cấu trúc truy vấn yêu cầu.

6. Projection, sắp xếp và giới hạn kết quả

from pymongo import DESCENDING

projection = {
    "_id": 0,
    "book_name": 1,
    "book_author": 1,
    "volume": 1,
}

cursor = (
    books.find({"volume": {"$gte": 1}}, projection)
    .sort("book_name", DESCENDING)
    .limit(10)
)

for book in cursor:
    pprint(book)

Projection có thể chọn các trường cần trả về hoặc loại bỏ các trường không cần thiết. Không trộn include và exclude trong cùng projection, ngoại trừ trường _id.

7. Cập nhật document

result = books.update_one(
    {"book_name": "Python Programming"},
    {"$set": {"volume": 4}},
)

print("Khớp:", result.matched_count)
print("Đã sửa:", result.modified_count)

matched_count cho biết số document khớp filter, còn modified_count cho biết số document thực sự thay đổi. Nếu muốn cập nhật tất cả document phù hợp, dùng update_many().

8. Xóa document

result = books.delete_one({"book_author": "David Beazley"})
print("Đã xóa:", result.deleted_count)

delete_one() chỉ xóa document đầu tiên khớp filter. delete_many({}) xóa toàn bộ document trong collection, vì vậy phải đặc biệt thận trọng và luôn kiểm tra filter trước.

9. Chương trình hoàn chỉnh

import os
from pprint import pprint
from pymongo import MongoClient, DESCENDING

uri = os.getenv("MONGODB_URI", "mongodb://localhost:27017/")

with MongoClient(uri, serverSelectionTimeoutMS=5000) as client:
    client.admin.command("ping")
    books = client["library"]["library_details"]

    for book in (
        books.find(
            {"volume": {"$gte": 2}},
            {"_id": 0, "book_name": 1, "book_author": 1, "volume": 1},
        )
        .sort("book_name", DESCENDING)
        .limit(10)
    ):
        pprint(book)

10. Bài thực hành

1. Tạo database Customer_purchase và collection Purchase_detail.
2. Chèn ba document có các trường Cust_id, Product_id, Cust_name, Purchase_product và Discount.
3. Tìm document có Discount bằng 0.5.
4. Liệt kê toàn bộ document nhưng ẩn _id.
5. Cập nhật Purchase_product của khách hàng Richard David thành Smartphone.
6. Thêm kiểm tra matched_count, modified_count và xử lý lỗi kết nối.

Tổng kết

Luồng PyMongo cơ bản gồm: tạo một MongoClient, kiểm tra kết nối, lấy database và collection, rồi gọi các phương thức CRUD. Hãy tái sử dụng client, giữ connection string trong biến môi trường, đọc các đối tượng kết quả và chỉ truy vấn những trường cần thiết để chương trình an toàn, rõ ràng và hiệu quả hơn.

Tham khảo: PyMongo Driver và CRUD Operations với PyMongo.