數據存儲：一文掌握存儲數據到MongoDB詳解

文章目錄

- 一、環境準備
- - 1.1 安裝MongoDB
  - 1.2 安裝Python MongoDB驅動
- 二、連接到MongoDB
- - 2.1 基本連接
  - 2.2 連接到MongoDB Atlas（云服務）
- 三、基本CRUD操作
- - 3.1 創建（Create）：插入數據
  - 3.2 讀取（Read）：查詢數據
  - 3.3 更新（Update）：修改數據
  - 3.4 刪除（Delete）：刪除數據
  - 3.5 完整示例
- 四、高級操作
- - 4.1 使用索引
  - 4.2 事務處理
  - 4.3 聚合管道
- 五、使用MongoEngine進行ODM操作
- - 5.1 安裝MongoEngine
  - 5.2 定義文檔模型
  - 5.3 執行CRUD操作
- 六、最佳實踐
- 七、總結

MongoDB是一種流行的NoSQL數據庫，以其靈活的文檔模型和高可擴展性而聞名。Python提供了多種庫來與MongoDB進行交互，其中最常用的是pymongo和mongoengine。本文將詳細介紹如何使用Python將數據存儲到MongoDB，包括環境設置、基本操作（CRUD）、高級功能以及最佳實踐。

關于mongo的更多樣例請參考：MongoDB基礎和進階

一、環境準備

1.1 安裝MongoDB

首先，確保已經在系統中安裝并運行了MongoDB。可以從 MongoDB官方網站下載并安裝適合你操作系統的版本。

1.2 安裝Python MongoDB驅動

使用pip安裝pymongo庫，這是Python中最常用的MongoDB驅動。

pip install pymongo

如果需要使用MongoDB的認證功能或其他高級特性，可以考慮安裝官方推薦的驅動：

pip install pymongo[srv]

二、連接到MongoDB

2.1 基本連接

以下是使用pymongo連接到MongoDB的基本示例：

from pymongo import MongoClient# 連接到本地MongoDB實例
client = MongoClient('mongodb://localhost:27017/')# 選擇數據庫，如果不存在則會在第一次寫入時創建
db = client['mydatabase']# 選擇集合（類似于表），如果不存在則會在第一次插入時創建
collection = db['mycollection']

2.2 連接到MongoDB Atlas（云服務）

如果使用MongoDB Atlas等云服務，連接字符串會包含用戶名和密碼：

from pymongo import MongoClient# 替換為你的連接字符串
uri = "mongodb+srv://<username>:<password>@cluster0.mongodb.net/mydatabase?retryWrites=true&w=majority"client = MongoClient(uri)
db = client['mydatabase']
collection = db['mycollection']

三、基本CRUD操作

3.1 創建（Create）：插入數據

插入單條文檔

document = {"name": "張三","age": 28,"email": "zhangsan@example.com"
}insert_result = collection.insert_one(document)
print(f"插入文檔的ID: {insert_result.inserted_id}")

插入多條文檔

documents = [{"name": "李四", "age": 34, "email": "lisi@example.com"},{"name": "王五", "age": 23, "email": "wangwu@example.com"}
]insert_many_result = collection.insert_many(documents)
print(f"插入文檔的IDs: {insert_many_result.inserted_ids}")

3.2 讀取（Read）：查詢數據

查詢所有文檔

results = collection.find()
for doc in results:print(doc)

條件查詢

query = {"age": {"$gt": 25}}
results = collection.find(query)
for doc in results:print(doc)

投影（選擇特定字段）

query = {"name": "張三"}
projection = {"_id": 0, "email": 1}
result = collection.find_one(query, projection)
print(result)

3.3 更新（Update）：修改數據

更新單個文檔

query = {"name": "李四"}
new_values = {"$set": {"age": 35}}
update_result = collection.update_one(query, new_values)
print(f"匹配的文檔數: {update_result.matched_count}, 修改的文檔數: {update_result.modified_count}")

更新多個文檔

query = {"age": {"$lt": 30}}
new_values = {"$inc": {"age": 1}}
update_result = collection.update_many(query, new_values)
print(f"匹配的文檔數: {update_result.matched_count}, 修改的文檔數: {update_result.modified_count}")

3.4 刪除（Delete）：刪除數據

刪除單個文檔

query = {"name": "王五"}
delete_result = collection.delete_one(query)
print(f"刪除的文檔數: {delete_result.deleted_count}")

刪除多個文檔

query = {"age": {"$gt": 30}}
delete_result = collection.delete_many(query)
print(f"刪除的文檔數: {delete_result.deleted_count}")

3.5 完整示例

from pymongo import MongoClientdef main():# 連接 MongoDBclient = MongoClient("mongodb://localhost:27017/")print("連接 MongoDB 成功")# 選擇數據庫和集合db = client["testdb"]collection = db["users"]print("數據庫和集合已選擇")try:# 插入單條數據data = {"name": "張三", "age": 25, "email": "zhangsan@example.com"}result = collection.insert_one(data)print(f"插入成功，文檔 ID: {result.inserted_id}")# 插入多條數據data = [{"name": "李四", "age": 30, "email": "lisi@example.com"},{"name": "王五", "age": 28, "email": "wangwu@example.com"}]result = collection.insert_many(data)print(f"插入成功，文檔 IDs: {result.inserted_ids}")# 查詢所有數據print("查詢所有數據:")results = collection.find()for result in results:print(result)# 條件查詢print("查詢 age 大于 25 的數據:")query = {"age": {"$gt": 25}}results = collection.find(query)for result in results:print(result)# 更新單條數據query = {"name": "張三"}new_values = {"$set": {"age": 26}}result = collection.update_one(query, new_values)print(f"更新成功，匹配文檔數: {result.matched_count}, 修改文檔數: {result.modified_count}")# 刪除單條數據query = {"name": "張三"}result = collection.delete_one(query)print(f"刪除成功，刪除文檔數: {result.deleted_count}")finally:# 關閉連接client.close()print("MongoDB 連接已關閉")if __name__ == "__main__":main()

運行上述代碼后，輸出如下：

連接 MongoDB 成功
數據庫和集合已選擇
插入成功，文檔 ID: 651f8b7e4b8e4e1f2c8e4f1a
插入成功，文檔 IDs: [651f8b7e4b8e4e1f2c8e4f1b, 651f8b7e4b8e4e1f2c8e4f1c]
查詢所有數據:
{'_id': ObjectId('651f8b7e4b8e4e1f2c8e4f1a'), 'name': '張三', 'age': 25, 'email': 'zhangsan@example.com'}
{'_id': ObjectId('651f8b7e4b8e4e1f2c8e4f1b'), 'name': '李四', 'age': 30, 'email': 'lisi@example.com'}
{'_id': ObjectId('651f8b7e4b8e4e1f2c8e4f1c'), 'name': '王五', 'age': 28, 'email': 'wangwu@example.com'}
查詢 age 大于 25 的數據:
{'_id': ObjectId('651f8b7e4b8e4e1f2c8e4f1b'), 'name': '李四', 'age': 30, 'email': 'lisi@example.com'}
{'_id': ObjectId('651f8b7e4b8e4e1f2c8e4f1c'), 'name': '王五', 'age': 28, 'email': 'wangwu@example.com'}
更新成功，匹配文檔數: 1, 修改文檔數: 1
刪除成功，刪除文檔數: 1
MongoDB 連接已關閉

四、高級操作

4.1 使用索引

索引可以提高查詢性能。以下是創建索引的示例：

# 創建單字段索引
collection.create_index("email", unique=True)# 創建復合索引
collection.create_index([("name", pymongo.ASCENDING), ("age", pymongo.DESCENDING)])

4.2 事務處理

MongoDB支持多文檔事務，適用于需要原子性的操作。

from pymongo import MongoClient
from pymongo.errors import ConnectionFailureclient = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['mycollection']with client.start_session() as session:with session.start_transaction():try:collection.insert_one({"name": "趙六", "age": 29}, session=session)collection.update_one({"name": "張三"}, {"$set": {"age": 28}}, session=session)except ConnectionFailure:session.abort_transaction()print("事務失敗，已回滾")

4.3 聚合管道

聚合管道用于數據處理和分析。

pipeline = [{"$match": {"age": {"$gt": 25}}},{"$group": {"_id": "$name", "total": {"$sum": 1}}},{"$sort": {"total": -1}}
]results = collection.aggregate(pipeline)
for doc in results:print(doc)

五、使用MongoEngine進行ODM操作

MongoEngine是一個基于MongoDB的對象文檔映射器（ODM），類似于Django ORM，適用于需要更高抽象層次的項目。

5.1 安裝MongoEngine

pip install mongoengine

5.2 定義文檔模型

from mongoengine import Document, StringField, IntField, connect# 連接到MongoDB
connect('mydatabase')class User(Document):name = StringField(required=True)age = IntField(required=True)email = StringField(unique=True)# 創建文檔
user = User(name="孫七", age=27, email="sunqi@example.com")
user.save()# 查詢文檔
user = User.objects(name="孫七").first()
print(user.email)

5.3 執行CRUD操作

# 更新
user.age = 28
user.save()# 刪除
user.delete()# 查詢所有
users = User.objects()
for u in users:print(u.name, u.age, u.email)

六、最佳實踐

?連接管理：使用連接池和上下文管理器來管理MongoDB連接，確保資源合理利用。

from pymongo import MongoClientwith MongoClient('mongodb://localhost:27017/') as client:db = client['mydatabase']# 執行操作

?索引優化：根據查詢需求創建合適的索引，避免全表掃描，提高查詢性能。

?數據驗證：使用MongoEngine等ODM框架進行數據驗證，確保數據一致性。

?安全性：

使用認證機制保護MongoDB實例。
避免在代碼中硬編碼敏感信息，使用環境變量或配置文件管理憑證。
?錯誤處理：捕獲并處理可能的異常（如使用pymongo.errors.PyMongoError ），確保應用的健壯性。

性能優化：對于大規模數據插入，可以使用 insert_many 的 ordered=False 參數來提高性能。

?合理使用事務：事務適用于需要原子性的多文檔操作，但不宜濫用，以免影響性能。

七、總結

使用Python存儲數據到MongoDB非常靈活且高效。通過pymongo庫，可以執行各種CRUD操作，并利用其豐富的功能滿足不同的需求。對于需要更高抽象層次的項目，MongoEngine提供了便捷的ODM支持。在實際應用中，結合最佳實踐，可以構建高性能、可維護的數據存儲解決方案。