初學者指南：全面了解 PyMongo 套件

更新日期： 2025 年 3 月 3 日

MongoDB 是現今最流行的 NoSQL 資料庫之一，而 PyMongo 則是 Python 連接 MongoDB 的首選工具。

如果你是一位剛開始接觸 MongoDB 與 Python 的開發者，這篇文章將帶你一步步了解 PyMongo，並教你如何在實際專案中使用它。

PyMongo 是什麼？

PyMongo 是 MongoDB 官方提供的 Python 資料庫 (Library)，它允許你在 Python 中與 MongoDB 資料庫進行溝通，包括資料的插入、查詢、更新與刪除 (CRUD 操作) 以及進階的聚合查詢 (Aggregation)。

PyMongo 具有以下特點：

• 簡單易用：提供了簡潔的 API，讓初學者也能快速上手。
• 高效能：支援批次操作與異步操作，適合處理大數據。
• 穩定可靠：作為官方工具，持續更新與維護。

為什麼選擇 PyMongo？

PyMongo 是 Python 連接 MongoDB 的官方客戶端，在眾多選擇中脫穎而出。

它不僅功能強大，而且易於學習和使用，特別適合需要處理大量數據的開發者。以下是 PyMongo 的幾大核心優勢：

1️⃣ 官方支持與持續更新

PyMongo 由 MongoDB 官方團隊開發和維護，這帶來了顯著的好處：

• 穩定性與安全性：官方開發的工具通常能夠保持高水準的品質和可靠性，在企業級應用中也能放心使用。
• 快速更新：PyMongo 能夠第一時間支援 MongoDB 最新版本和功能，讓開發者隨時掌握最新技術動向。
• 完善的技術支持：擁有豐富的官方文件、範例代碼和技術社群，解決開發中遇到的問題更加便捷。

2️⃣ 功能豐富，應用場景廣泛

PyMongo 支援各種資料庫操作，無論是簡單的數據操作還是複雜的數據分析，它都能應付自如：

• 完整的 CRUD 操作：提供 insert_one、find、update_many、delete_one 等方法，靈活處理數據插入、查詢、更新和刪除。
• 聚合查詢 (Aggregation)：內建強大的聚合管道，適合進行數據分組、過濾和統計分析，讓 MongoDB 成為數據分析工具的一部分。
• 索引操作 (Indexing)：可以輕鬆建立單欄位或複合索引 (create_index)，顯著提升數據查詢效能。
• 批次操作 (Bulk Operations)：支援批量插入、更新和刪除操作 (bulk_write)，在大數據應用中尤為重要。
• 事務 (Transactions) 支持：可確保多步資料庫操作的原子性和一致性，特別適合複雜的商務應用場景。

3️⃣ 符合 Python 編程習慣，學習曲線平緩

PyMongo 的 API 設計十分 Pythonic，初學者也能快速上手：

• 直觀的語法：PyMongo 中的資料操作大多以 Python 字典和列表進行，查詢語法與 MongoDB Shell 類似，例如：

result = collection.find({"name": "John", "age": {"$gt": 25}})

• 易於結合其他工具：可以與 Python 其他熱門框架和工具無縫整合，例如：
  • Pandas：將查詢結果轉換成 DataFrame，便於數據分析。
  • Flask/Django：快速開發基於 MongoDB 的 Web 應用程式。
  • FastAPI + Motor：實現高效能的異步 API 服務，適合高併發場景。
• 靈活性高：PyMongo 允許使用原生 MongoDB 語法，同時提供高階 API，讓開發者可以根據需求選擇不同的操作方式。

4️⃣ 高效能與擴展性

PyMongo 在應對大型應用和高流量情況時依然能保持良好效能：

• 連接池 (Connection Pooling)：內建連接池機制，避免頻繁建立和關閉資料庫連接，顯著提升應用效能。
• 分片 (Sharding) 支援：當數據量增大時，PyMongo 可以與 MongoDB 分片技術結合，自動將數據分佈到多個伺服器，確保系統高可擴展性。
• 效能調優選項：提供讀取偏好 (Read Preference) 和寫入一致性 (Write Concern) 設置，讓開發者能夠在數據一致性與系統效能之間靈活平衡。

5️⃣ 強大的錯誤處理與日誌系統

• 全面的異常處理：PyMongo 提供了豐富的錯誤類別 (pymongo.errors)，如連接超時、操作失敗、伺服器錯誤等，幫助開發者進行精細的錯誤處理：

from pymongo import errors

try:
    collection.insert_one({"name": "John"})
except errors.DuplicateKeyError:
    print("該數據已存在，無法插入！")
except errors.ConnectionFailure:
    print("無法連接到資料庫！")

• 日誌支持：結合 Python 的 logging 模組，PyMongo 可以記錄數據操作和錯誤信息，有助於系統監控和問題排查。

PyMongo 的安裝與設定

環境準備

在開始之前，請確認你的開發環境中已經安裝了以下工具：

• Python 版本：3.7 或更高版本
• pip 套件管理器

安裝 PyMongo

你可以使用 pip 來安裝 PyMongo：

pip install pymongo

若需要特定版本，請使用以下命令：

pip install pymongo==4.5.0

安裝 MongoDB

若你本地還沒有安裝 MongoDB，可以參考 官方文件 進行安裝，或是選擇使用 MongoDB Atlas 這類的雲端資料庫。

基本操作教學

在學習 PyMongo 的過程中，掌握基本的資料庫操作是最核心的部分。本節將詳細介紹如何使用 PyMongo 與 MongoDB 進行連接，並執行資料的插入、查詢、更新和刪除 (CRUD) 操作。

連接 MongoDB 資料庫

在使用 PyMongo 之前，首先需要與 MongoDB 資料庫建立連接。這可以通過 MongoClient 來實現。

from pymongo import MongoClient

# 建立 MongoDB 連接
client = MongoClient("mongodb://localhost:27017/")
db = client["mydatabase"]  # 建立或選擇資料庫

🛠️ 代碼解析：

• MongoClient：MongoClient 是 PyMongo 提供的連接 MongoDB 的客戶端，透過它可以與資料庫伺服器進行通訊。
• 連接字串："mongodb://localhost:27017/" 是連接字串的格式：
  • localhost：表示伺服器位址 (本地端)；若使用遠端伺服器，需將此處替換為對應的 IP 或域名。
  • 27017：MongoDB 的預設埠號 (Port)，可根據實際設定調整。
• 資料庫選擇：db = client["mydatabase"] 用於選擇資料庫 (Database)。如果資料庫不存在，MongoDB 會在插入資料時自動創建它。

資料庫與集合的建立

在 MongoDB 中，資料庫 (Database) 包含一組集合 (Collection)，而集合則類似於關聯數據庫中的資料表 (Table)。

collection = db["customers"]  # 建立或選擇集合

📂 集合 (Collection) 是什麼？

• 集合 (Collection)：是一組文件 (Document) 的集合，在 MongoDB 中，每個文件都是一個類似 JSON 的 BSON (Binary JSON) 結構。
• 舉例來說，"customers" 就像一個專門存儲客戶數據的資料表，未來可以插入、查詢、更新和刪除與客戶相關的資料。
• 如果指定的集合名稱不存在，MongoDB 會在插入資料時自動創建集合，這讓開發流程更加靈活。

資料插入 (Insert)

插入資料是將新的文件新增到集合中的操作，PyMongo 支援單筆和多筆資料的插入。

🧾 插入單筆資料

# 單筆插入範例
customer = {"name": "John", "age": 25, "city": "New York"}
result = collection.insert_one(customer)

# 查看插入結果
print("插入的文檔 ID:", result.inserted_id)

• insert_one()：將單一字典格式的資料插入集合。
• result.inserted_id：返回插入資料後自動生成的 _id 欄位，這是一個唯一識別值 (ObjectId)，方便後續查詢和管理。

📑 插入多筆資料

# 插入多筆資料
customers = [
    {"name": "Anna", "age": 22},
    {"name": "Mike", "age": 32}
]
result = collection.insert_many(customers)

# 查看所有插入的文檔 ID
print("插入的多筆文檔 ID:", result.inserted_ids)

• insert_many()：接受一個列表 (List)，列表中的每個元素都是一個字典 (Document)。
• 批量插入：特別適合初始化資料或批次導入數據的情境。

資料查詢 (Query)

查詢資料是從集合中讀取符合條件的文件，可以使用 find() 和 find_one() 方法進行。

🔍 查詢所有資料

# 迴圈遍歷集合中的所有文檔
for doc in collection.find():
    print(doc)

• collection.find()：無參數時會返回集合中的所有資料，結果以迭代器 (Cursor) 形式返回，需使用迴圈遍歷。

🎯 查詢特定條件資料

# 查詢姓名為 "John" 的文檔
result = collection.find_one({"name": "John"})
print("找到的文檔:", result)

• find_one()：只返回第一個匹配的文檔，如果沒有匹配，返回 None。
• 查詢條件使用字典格式，例如 {"name": "John"} 表示查找 "name" 欄位等於 "John" 的文件。

資料更新 (Update)

資料更新操作允許你修改集合中的現有文檔，常用的方法有 update_one() 和 update_many()。

✏️ 單筆資料更新

# 將 "John" 的年齡更新為 30 歲
collection.update_one({"name": "John"}, {"$set": {"age": 30}})

• $set 操作符：用於修改指定欄位的值，其他欄位保持不變。
• 第一個參數：指定查詢條件 (如 {"name": "John"})；
• 第二個參數：定義更新的內容 (如 {"$set": {"age": 30}})。

🧮 多筆資料更新

# 將年齡小於 30 的客戶，城市更新為 "San Francisco"
collection.update_many({"age": {"$lt": 30}}, {"$set": {"city": "San Francisco"}})

• update_many()：用於批次更新所有符合條件的文檔，特別適合需要大範圍數據修改的場景。
• 比較操作符 $lt：表示小於 (less than)，MongoDB 支援多種查詢操作符 (如 $gt、$eq、$in 等)。

資料刪除 (Delete)

刪除操作允許你從集合中移除不需要的數據，PyMongo 提供了 delete_one() 和 delete_many() 方法。

❌ 刪除特定資料

# 刪除姓名為 "John" 的文檔
collection.delete_one({"name": "John"})

• delete_one()：僅刪除第一個符合條件的文檔，即使有多筆數據符合條件。

🚮 刪除多筆資料

# 刪除城市為 "San Francisco" 的所有文檔
collection.delete_many({"city": "San Francisco"})

• delete_many()：用於批次刪除所有符合條件的文檔，在清理數據或刪除過期數據時非常有用。

進階功能

在掌握了 PyMongo 的基本資料庫操作後，我們還可以利用一些進階功能，讓資料操作更加高效且具有分析能力。

本節將介紹批次操作 (Bulk Operations)、索引 (Indexing) 和聚合操作 (Aggregation)，這些功能能幫助你處理大數據場景，提升應用效能和數據處理能力。

批次操作 (Bulk Operations)

批次操作允許開發者在單一請求中，執行多個資料庫操作，這在處理大量數據或需要同時進行多種操作時非常有用。

例如，可以同時插入、更新和刪除多個文檔 (Documents)，避免多次與資料庫通訊，從而顯著提升效能。

🚀 為什麼需要批次操作？

• 效能提升：批次操作只需一次請求即可完成多個操作，減少資料庫連接次數，降低網路延遲。
• 事務性操作：可以確保一組操作要麼全部成功，要麼全部失敗，維持數據一致性。
• 複雜場景應用：適合初始化數據、數據遷移、批次修改數據等場景。

🛠️ 批次操作示例

from pymongo import InsertOne, DeleteOne, UpdateOne

# 建立批次操作請求列表
requests = [
    InsertOne({"name": "Chris", "age": 20}),  # 插入新資料
    DeleteOne({"name": "Anna"}),              # 刪除符合條件的資料
    UpdateOne({"name": "Mike"}, {"$set": {"age": 35}})  # 更新現有資料
]

# 執行批次操作
result = collection.bulk_write(requests)

# 查看批次操作結果
print(result.bulk_api_result)

🔍 代碼解析

• 操作類型：
  • InsertOne()：插入新文件，類似於 insert_one()。
  • DeleteOne()：刪除符合條件的第一個文件，與 delete_one() 類似。
  • UpdateOne()：更新第一個符合條件的文件，功能與 update_one() 相同。
• bulk_write() 方法：
  • 接受一個操作請求的列表，這些請求可以是不同類型的數據操作。
  • 操作的順序按照列表順序執行，因此可以精確控制操作流程。
• bulk_api_result：返回一個字典，包含了所有操作的執行情況，例如插入、更新、刪除的數量：

{
    'nInserted': 1,
    'nUpserted': 0,
    'nMatched': 1,
    'nModified': 1,
    'nRemoved': 1
}

• nInserted：插入的文檔數量。
• nMatched：匹配到的文檔數量 (適用於更新操作)。
• nModified：實際被更新的文檔數量。
• nRemoved：刪除的文檔數量。

📈 使用場景

• 數據初始化：在應用啟動時一次性插入大量預設數據。
• 數據同步：在數據庫間同步數據時，將新增、更新、刪除操作組合成一個批次請求。
• 數據清理：批量刪除過期或不需要的數據，提高資料庫效能。

索引 (Indexing)

索引在資料庫中就像是書籍的目錄，能夠加速數據的查詢速度。

在 MongoDB 中，索引對於大型集合中的複雜查詢尤為重要，尤其是需要經常搜索特定欄位的情況下。

🚦 為什麼要使用索引？

• 提高查詢效能：建立索引後，查詢數據時資料庫不需要遍歷所有文檔，從而顯著縮短查詢時間。
• 支援排序操作：索引不僅加速查詢，還能提升排序操作 (如 sort()) 的效能。
• 保證欄位唯一性：透過唯一索引 (unique=True)，可以避免資料重複插入，例如用於 Email 或用戶名的唯一性約束。

🔍 建立索引的示例

# 為 "name" 欄位建立升冪排序的索引
collection.create_index([("name", 1)])  # 1 表示升冪 (Ascending)，-1 表示降冪 (Descending)

• 複合索引：可以對多個欄位建立索引，例如：

# 對 "name" 和 "age" 欄位建立複合索引，"name" 升冪，"age" 降冪
collection.create_index([("name", 1), ("age", -1)])

• 唯一索引：防止插入重複資料：

collection.create_index([("email", 1)], unique=True)

聚合操作 (Aggregation)

聚合操作是數據分析和統計的重要工具。它允許開發者對數據進行複雜的處理，例如分組、計算平均值、篩選特定條件等，功能類似於 SQL 中的 GROUP BY 和 JOIN。

📊 聚合管道 (Aggregation Pipeline)

聚合操作透過一系列的 “管道階段” (Pipeline Stages) 來逐步處理數據，每個階段進行特定的操作，最終輸出結果。

pipeline = [
    {"$match": {"age": {"$gte": 25}}},  # 過濾年齡大於等於 25 的文檔
    {"$group": {"_id": "$city", "average_age": {"$avg": "$age"}}}  # 按城市分組，計算平均年齡
]

# 執行聚合操作
for result in collection.aggregate(pipeline):
    print(result)

🔍 代碼解析

• $match：類似於查詢操作，用於篩選符合條件的文檔，例如篩選年齡大於等於 25 的用戶。
• $group：將數據按指定欄位分組，_id 代表分組的依據，$avg 則是計算每組的平均值。
• 結果示例：

{"_id": "New York", "average_age": 28.5}
{"_id": "San Francisco", "average_age": 30.0}

🌟 常見的聚合操作

• $sum：計算總和。
• $avg：計算平均值。
• $min / $max：計算最小值和最大值。
• $sort：對聚合結果進行排序。
• $limit：限制結果集數量。

注意事項與最佳實踐

在使用 PyMongo 與 MongoDB 開發應用程式時，不僅要掌握各種操作方法，還需要注意效能、穩定性和數據的一致性。

本節將介紹幾個關鍵的注意事項與最佳實踐，包括連接池、異常處理和資料驗證，這些方法能夠幫助你避免常見問題，確保應用程式運行順暢。

連接池 (Connection Pooling)

💡 什麼是連接池？

連接池 (Connection Pooling) 是一種管理資料庫連接的技術，它允許應用程式在初始化時創建一組資料庫連接，並在應用程序的整個生命周期內重複使用這些連接。

而不是每次需要訪問資料庫時，都重新建立和關閉連接。

🚦 為什麼要使用連接池？

• 效能提升：建立資料庫連接需要耗費時間 (TCP 連接、身份驗證等)，重複使用連接能顯著減少延遲。
• 減少資源消耗：避免頻繁建立和關閉連接，有助於減少伺服器和網路資源的開銷。
• 提高穩定性：對於高併發應用，連接池能有效控制同時連接的數量，避免伺服器過載。

🛠️ PyMongo 中的連接池示例

在 PyMongo 中，MongoClient 預設已經啟用了連接池功能，但你可以進一步自訂連接池的參數：

from pymongo import MongoClient

# 建立連接池，最大連接數為 50，最小空閒連接數為 10
client = MongoClient(
    "mongodb://localhost:27017/",
    maxPoolSize=50,  # 最大連接數
    minPoolSize=10,  # 最小空閒連接數
    serverSelectionTimeoutMS=5000  # 伺服器連接超時 (5 秒)
)

db = client["mydatabase"]

🔍 參數解釋

• maxPoolSize：連接池中允許的最大連接數量，默認為 100。
• minPoolSize：當連接池空閒時保持的最小連接數，預設為 0。
• serverSelectionTimeoutMS：伺服器選擇超時設定，避免連接失敗時應用程式無限等待。

🚧 注意事項

• 長時間空閒連接的管理：如果你的應用程式中有長時間不使用資料庫連接的情況，建議定期測試連接的有效性，或者設定 socketTimeoutMS 確保空閒連接不會被無限期占用。
• 在應用關閉時關閉客戶端：確保在程式結束時關閉 MongoClient，避免資源洩漏：

client.close()

異常處理 (Error Handling)

在與資料庫進行通訊時，難免會遇到各種錯誤，例如：連接超時、數據格式錯誤、操作失敗等。

良好的異常處理機制不僅能避免應用程式崩潰，還能幫助快速定位問題。

🛠️ PyMongo 異常處理示例

PyMongo 提供了一組豐富的錯誤類別 (pymongo.errors)，可以透過 try...except 語句進行捕獲和處理。

from pymongo import errors

try:
    # 插入資料範例
    collection.insert_one({"name": "John", "age": 30})

except errors.ConnectionFailure as e:
    print("無法連接到 MongoDB 伺服器:", e)

except errors.DuplicateKeyError as e:
    print("重複的鍵值錯誤:", e)

except errors.OperationFailure as e:
    print("資料庫操作失敗:", e)

except Exception as e:
    print("發生未預期的錯誤:", e)

🔍 常見錯誤類型

• ConnectionFailure：無法連接到資料庫，可能是伺服器關閉或網路問題。
• DuplicateKeyError：當插入的文檔違反唯一索引 (如 _id 或 unique=True 的欄位) 時引發。
• OperationFailure：操作失敗，通常發生在無效的查詢或權限不足時。
• TimeoutError：資料庫請求超時，例如查詢操作超過設定的時間限制。

🚧 最佳實踐

• 日誌記錄 (Logging)：在捕獲錯誤時，將錯誤訊息記錄到日誌中，以便後續分析問題：

import logging

logging.basicConfig(filename='app.log', level=logging.ERROR)

try:
    collection.find_one({"name": "John"})
except Exception as e:
    logging.error("資料庫錯誤: %s", e)

資料驗證 (Data Validation)

在將資料插入 MongoDB 之前，進行資料格式和類型的驗證可以防止不正確或不一致的數據進入資料庫，保持數據的整潔性和可靠性。

📋 為什麼需要資料驗證？

• 防止錯誤數據污染資料庫：特別是在多用戶應用中，避免因用戶輸入錯誤導致數據格式混亂。
• 提升資料庫查詢的準確性：保持數據結構一致，能避免查詢語句失敗或返回錯誤結果。

🛠️ 使用 Pydantic 進行資料驗證

Pydantic 是一個用於數據驗證的 Python 庫，特別適合與 PyMongo 結合使用，能自動驗證數據格式，並提供自動轉換和錯誤提示。

from pydantic import BaseModel, Field, ValidationError

# 定義數據模型
class CustomerModel(BaseModel):
    name: str
    age: int = Field(..., ge=0, le=120)  # 年齡需在 0 到 120 歲之間
    city: str

# 驗證輸入資料
try:
    customer = CustomerModel(name="John", age=25, city="New York")
    collection.insert_one(customer.dict())  # 驗證通過後插入資料庫

except ValidationError as e:
    print("資料格式錯誤:", e)

🔍 Pydantic 的優勢

• 類型驗證：自動檢查欄位的數據類型是否正確 (str, int, float 等)。
• 欄位限制：透過 Field 設定更精細的限制，例如範圍、最小/最大長度等。
• 錯誤信息清晰：在驗證失敗時提供詳細的錯誤資訊，方便開發者調試。

總結

PyMongo 是一個功能強大且易於使用的 MongoDB 客戶端，無論是基本的 CRUD 操作，還是進階的聚合與索引功能，它都能幫助我們高效地處理資料。

在掌握了這些基礎之後，你可以在自己的專案中自由運用 PyMongo，建立更靈活且高效的資料管理系統。