pymongo庫：簡易方式存取數據

文檔

基礎使用

前提：開發機器已安裝mongo配置環境，已啟動服務。
macOS啟動服務：brew services start mongodb-community@8.0
macOS停止服務：brew services stop mongodb-community@8.0
安裝：python3 -m pip install pymongo
導入：import pymongo

與MongoClient建立連接：client = MongoClient()

from pymongo import MongoClient# 創建連接的三種方式：（任選一種即可）
## 第一種：使用默認的主機 localhost 和端口號 27017 創建連接
client = MongoClient()
## 第二種：指定主機和端口創建連接
client_1 = MongoClient("localhost", 27017)
## 第三種：使用url格式創建連接
client_2 = MongoClient("mongodb://localhost:27017/")# 有賬號密碼時的連接
client3 = MongoClient(f"mongodb://{username}:{password}@localhost:27017/mydatabase?authSource=admin")
## 或者使用獨立參數
client4 = MongoClient(host='localhost',port=27017,username='your_username',password='your_password@123',authSource='admin',  # 認證數據庫
)
## 若密碼包含特殊字符，則需要額外處理
from urllib.parse import quote_plus
safe_password = quote_plus("p@ssw:rd/$")

獲取數據庫：db = client.test_database

# 獲取數據庫的兩種方式：
## 第一種：通過 .數據庫名 獲取
db = client.test_database
## 第二種：如果數據庫名含特殊字符，不能使用屬性樣式訪問時，可通過 字典式 訪問
db1 = client['test-database']

收集（數據庫表）：collection = db.test_collection

# 獲取數據庫表的兩種方式：
## 第一種：通過 .表名 獲取
collection = db.test_collection
## 第二種：如果表名含特殊字符，不能使用屬性樣式訪問時，可通過 字典式 訪問
collection1 = db['test-collection']

文件（表數據）：可以包含原生Python類型(如 datetime.datetime 實例)，它們將自動轉換為相應的 BSON 類型。（二進制json格式）

# 構造表數據
import datetime
post = {"author": "Mike","text": "My first blog post!","tags": ["mongodb", "python", "pymongo"],"date": datetime.datetime.now(tz=datetime.timezone.utc),
}

插入文檔（將數據插入到數據庫表中）：collection.insert_one(post)
??注意：由于延遲創建，此時（在向集合和數據庫中插入第一個文檔時）才會創建這些集合和數據庫。

# 將文檔插入到集合中（即將數據插入到數據庫表）
post_id = collection.insert_one(post).inserted_id
print(post_id)  # 插入數據對應的id
# 查看集合（數據庫表）
db.list_collection_names()  # ['collection']

獲取單個文檔（獲取表中某個數據）：result = collection.find_one({'key':'value'})
??注意：返回的文檔包含 “_id” ，在插入時自動添加。（也可以按{'_id': ObjectId(post_id)}來查找）
```
# 默認獲取第一個數據
collection.find_one()
# 根據條件查找匹配的數據
collection.find_one({'key':'value'})
```

大容量插入（批量插入多條數據）：result = collection.insert_many(collection_data_list)

new_posts = [{"author": "Mike","text": "Another post!","tags": ["bulk", "insert"],"date": datetime.datetime(2009, 11, 12, 11, 14),},{"author": "Eliot","title": "MongoDB is fun","text": "and pretty easy too!","date": datetime.datetime(2009, 11, 10, 10, 45),},
]
result = collection.insert_many(new_posts)  # 參數為列表
result.inserted_ids  # 可以獲取插入后的id列表

查詢多個文檔：result = collection.find({'key':'value'})，result是在collection表內搜索key=value的結果列表。
計數：
- 獲取所有數據條數：collection.count_documents({})
- 獲取特定條件的數據條數：collection.count_documents({"author": "Mike"})
范圍查詢：collection.find({"date": {"$lt": date_value}}).sort("author")，查詢date_value日期之前的帖子，按作者對結果進行排序。
```
d = datetime.datetime(2009, 11, 12, 12)
result = collection.find({"date": {"$lt": d}}).sort("author")  # result是一個字典列表
```

索引：添加索引有助于加速某些查詢，還可以為查詢和存儲文檔添加附加功能。

result = collection.create_index([("user_id", pymongo.ASCENDING)], unique=True)  # 將user_id設置為唯一索引
indexs = sorted(list(collection.index_information()))  # 獲取collection表中已創建的索引列表

PyMongo 常用方法匯總

方法	使用場景	示例代碼	關鍵特性
`find()`	條件查詢、數據檢索	`db.users.find({"status": "active"}, {"name": 1, "email": 1})`	支持投影、排序、分頁鏈式操作
`insert_one()`	插入單個文檔	`result = db.products.insert_one({"name": "Mouse", "price": 29.99})` `print(result.inserted_id)`	返回插入文檔的_id
`insert_many()`	批量插入文檔	`users = [{"name": "Alice"}, {"name": "Bob"}]` `result = db.users.insert_many(users, ordered=False)`	支持有序/無序插入，返回插入ID列表
`update_one()`	更新單個文檔	`db.users.update_one(` `{"_id": 123},` `{"$set": {"status": "active"}}` `)`	使用$set避免覆蓋整個文檔
`update_many()`	批量更新文檔	`result = db.orders.update_many(` `{"status": "pending"},` `{"$inc": {"retries": 1}}` `)`	返回匹配和修改的文檔計數
`delete_one()`	刪除單個文檔	`db.sessions.delete_one({"expire_at": {"$lt": datetime.now()}})`	刪除第一個匹配文檔
`delete_many()`	批量刪除文檔	`result = db.logs.delete_many({"created": {"$lt": datetime(2023,1,1)}})` `print(result.deleted_count)`	返回刪除的文檔數量
`count_documents()`	統計符合條件的文檔數量	`active_users = db.users.count_documents(` `{"status": "active", "last_login": {"$gt": last_month}}` `)`	替代已廢棄的count()方法
`distinct()`	獲取字段的唯一值列表	`categories = db.products.distinct("category")`	比聚合$group更高效
`aggregate()` ??	復雜數據分析、多階段處理	`pipeline = [` `{"$match": {"status": "active"}},` `{"$group": {"_id": "$dept", "count": {"$sum": 1}}}` `]` `results = db.users.aggregate(pipeline)`	支持 `$match`, `$group`, `$lookup`等聚合階段，先 `$match`過濾比較高效
`find_one()`	查找單個文檔	`user = db.users.find_one({"email": "user@example.com"})`	直接返回文檔對象或None
`find_one_and_update()`	原子查找并更新文檔	`task = db.tasks.find_one_and_update(` `{"status": "pending"},` `{"$set": {"status": "processing"}},` `return_document=ReturnDocument.AFTER` `)`	保證操作的原子性
`create_index()`	創建查詢索引	`db.orders.create_index([("user_id", 1), ("create_time", -1)])`	顯著提升查詢性能
`bulk_write()`	高性能批量操作	`ops = [` `UpdateOne({"id": 1}, {"$inc": {"views": 1}}),` `DeleteMany({"expired": True})` `]` `result = db.collection.bulk_write(ops, ordered=False)`	減少網絡往返，提升吞吐量，`ordered=False`無序操作更快
`drop()`	刪除整個集合	`db.temp_data.drop()`	謹慎使用，不可逆操作

根據使用場景選擇對應的方法：
在這里插入圖片描述

aggregate中pipeline的核心聚合操作符速查表

操作符	類別	功能描述	語法示例
`$match`	篩選階段	過濾文檔 (類似SQL的WHERE)	`{"$match": {"status": "active"}}`
`$project`	重塑階段	選擇/重命名字段 (類似SELECT)	`{"$project": {"name": 1, "year": {"$year": "$date"}}}`
`$group`	分組階段	按字段分組并計算聚合值	`{"$group": {"_id": "$dept", "total": {"$sum": "$salary"}}}`
`$sort`	排序階段	結果排序 (1升序, -1降序)	`{"$sort": {"age": -1, "name": 1}}`
`$limit`	限制階段	限制輸出文檔數量	`{"$limit": 10}`
`$skip`	跳過階段	跳過指定數量文檔	`{"$skip": 5}`
`$unwind`	數組處理	展開數組為多條文檔	`{"$unwind": "$tags"}`
`$lookup`	關聯查詢	跨集合關聯查詢 (類似SQL JOIN)	`{"$lookup": {` `"from": "products", # 關聯集合` `"localField": "product_id", # 本地字段` `"foreignField": "_id", # 關聯集合字段` `"as": "product_info" # 輸出字段名}}`
`$addFields`	字段操作	添加新字段 (不改變原字段)	`{"$addFields": {"discount": {"$multiply": ["$price", 0.9]}}}`
`$count`	統計階段	返回文檔總數	`{"$count": "total_users"}`
`$facet`	多管道處理	同一輸入執行多個聚合管道	`pipeline = [ # 單次查詢獲取多種統計結果` `{"$facet": {` `"department_stats": [ # 部門統計管道` `{"$group": {... }}],` `"age_distribution": [ # 年齡分布管道` `{"$bucket": {... }} ],` `"total_count": [ # 總計數管道` `{"$count": "value"} ]` `}}]`

在聚合管道中使用的核心表達式操作符：

類型	操作符
算術運算	`$add`, `$subtract`, `$multiply`, `$divide`, `$mod`
比較運算	`$eq`, `$ne`, `$gt`, `$gte`, `$lt`, `$lte`, `$cmp`
邏輯運算	`$and`, `$or`, `$not`, `$cond` (三元表達式)
日期處理	`$year`, `$month`, `$dayOfMonth`, `$hour`, `$minute`, `$dateToString`
字符串處理	`$concat`, `$substr`, `$toLower`, `$toUpper`, `$trim`, `$split`
數組處理	`$size`, `$slice`, `$map`, `$filter`, `$in`, `$arrayElemAt`
數據類型	`$type`, `$convert`, `$toInt`, `$toString`, `$toDate`
條件處理	`$ifNull`, `$switch` (多分支條件)

聚合管道優化策略

管道順序優化原則
性能優化技巧
- 將 $match 和 $project 放在管道最前端減少數據處理量
- 在 $lookup 前使用 $match 過濾關聯集合數據
- 避免在 $group 中使用 $push 操作大數組
- 對 $sort 和 $match 使用的字段創建索引
- 使用 allowDiskUse=True 處理大數據集
```
 result = collection.aggregate(pipeline, allowDiskUse=True)
```