1. 安装 pymongo

如果没有安装pymongo 库，可以通过以下命令进行安装：

pip install pymongo

2. 连接 MongoDB

在开始操作之前，需要连接到 MongoDB 数据库，可以使用 pymongo 提供的 MongoClient 类来连接到本地或远程的 MongoDB 实例

from pymongo import MongoClient, ReadPreference# 连接到本地的 MongoDB 实例
## client = MongoClient('mongodb://localhost:27017/')
mongo_uri = 'mongodb://user:pwd@localhost:27017/admin'
client = MongoClient(mongo_uri, read_preference=ReadPreference.SECONDARY)# 连接到远程 MongoDB 实例（例如，使用 MongoDB Atlas）
# client = MongoClient('mongodb+srv://<username>:<password>@cluster0.mongodb.net/myFirstDatabase?retryWrites=true&w=majority')# 选择数据库
db = client['mydatabase']# 选择集合
collection = db['mycollection']

3. 创建（插入）文档

MongoDB 中的文档是 JSON 风格的 BSON（Binary JSON）格式，可以使用 insert_one() 和 insert_many() 方法插入单个或多个文档

插入单个文档

# 插入单个文档
document = {"name": "ZhangSan","age": 25,"skills": ["Python", "MongoDB"],"address": {"street": "123 Main St","city": "Macau","zip": "10001"}
}result = collection.insert_one(document)
print("Inserted document ID:", result.inserted_id)

插入多个文档

# 插入多个文档
documents = [{"name": "LiSi","age": 30,"skills": ["Java", "MongoDB"],"address": {"street": "456 Elm St","city": "Chicago","zip": "60601"}},{"name": "WangWu","age": 35,"skills": ["JavaScript", "React"],"address": {"street": "789 Oak St","city": "San Francisco","zip": "94101"}}
]result = collection.insert_many(documents)
print("Inserted document IDs:", result.inserted_ids)

4. 查询文档

MongoDB 提供了丰富的查询功能，可以使用 find_one() 和 find() 方法进行查询

查询单个文档

# 查询单个文档
document = collection.find_one({"name": "Alice"})
print(document)

查询多个文档

# 查询多个文档
cursor = collection.find({"age": {"$gt": 25}})
for doc in cursor:print(doc)

复杂查询

MongoDB 支持复杂的查询操作符，如 $or, $and, $in, $gt, $lt 等

# 复杂查询：查询年龄大于 25 且技能包含 "MongoDB" 的文档
query = {"age": {"$gt": 25},"skills": "MongoDB"
}cursor = collection.find(query)
for doc in cursor:print(doc)

嵌套查询

MongoDB 支持嵌套文档的查询

# 查询地址城市为 "Macau" 的文档
query = {"address.city": "Macau"
}cursor = collection.find(query)
for doc in cursor:print(doc)

分页条件查询（通用模版）

t_table_name = "t_test" # 目标表名
page_size = 2000 # 单次迭代查询页数
cond = {"name":"ZhangSan"} #查询条件，可为空
sort = [("_id", 1)] # 排序条件
cond["address.city"] = {"$eq": "Macau"} #嵌套文档查询条件
num = 0
dealing = False # 用于标识查询处理是否完毕
projection = {"name":1, "age":1} # 过滤输出目标字段，例如只输出name和age字段，可为空，空表示输出完整文档
while(True):for item in log_db[t_table_name].find(cond, sort = sort, projection = projection, limit = page_size):dealing = Truenum = num + 1cond["_id"] = {"$gt": item["_id"]}# TODO 处理业务逻辑if not dealing:breakprint("finish", num) # 统计查询到的数量dealing = False

5. 更新文档

MongoDB 提供了 update_one() 和 update_many() 方法来更新文档。可以使用 $set, $inc, $push, $pull 等更新操作符

更新单个文档

# 更新单个文档
query = {"name": "ZhangSan"}
new_values = {"$set": {"age": 26}}result = collection.update_one(query, new_values)
print("Matched count:", result.matched_count)
print("Modified count:", result.modified_count)

更新多个文档

# 更新多个文档
query = {"age": {"$lt": 35}}
new_values = {"$inc": {"age": 1}}  # 将年龄加 1result = collection.update_many(query, new_values)
print("Matched count:", result.matched_count)
print("Modified count:", result.modified_count)

更新嵌套文档

# 更新嵌套文档
query = {"name": "Alice"}
new_values = {"$set": {"address.city": "Los Angeles"}}result = collection.update_one(query, new_values)
print("Matched count:", result.matched_count)
print("Modified count:", result.modified_count)

6. 删除文档

MongoDB 提供了 delete_one() 和 delete_many() 方法来删除文档

删除单个文档

# 删除单个文档
query = {"name": "ZhangSan"}result = collection.delete_one(query)
print("Deleted count:", result.deleted_count)

删除多个文档

# 删除多个文档
query = {"age": {"$gt": 30}}result = collection.delete_many(query)
print("Deleted count:", result.deleted_count)

7. 处理复杂的文档结构

MongoDB 支持非常灵活的文档结构，可以嵌套数组、嵌套对象等

插入带有数组的文档

# 插入带有数组的文档
document = {"name": "David","age": 40,"skills": ["Python", "MongoDB", "Data Science"],"projects": [{"name": "Project A","status": "Completed"},{"name": "Project B","status": "In Progress"}]
}result = collection.insert_one(document)
print("Inserted document ID:", result.inserted_id)

查询嵌套数组中的元素

# 查询项目名为 "Project A" 的文档
query = {"projects.name": "Project A"
}cursor = collection.find(query)
for doc in cursor:print(doc)

更新嵌套数组中的元素

# 更新项目状态为 "Completed" 的文档
query = {"projects.name": "Project B"
}new_values = {"$set": {"projects.$[elem].status": "Completed"  # 使用数组过滤器}
}update_result = collection.update_many(query, new_values, array_filters=[{"elem.name": "Project B"}])
print("Matched count:", update_result.matched_count)
print("Modified count:", update_result.modified_count)

8. 批量操作

MongoDB 支持批量操作，可以提高性能。可以使用 bulk_write() 方法进行批量插入、更新、删除等操作

from pymongo import InsertOne, UpdateOne, DeleteOne# 批量操作
requests = [InsertOne({"name": "Eve","age": 29,"skills": ["Python", "JavaScript"]}),UpdateOne({"name": "Bob"},{"$set": {"age": 31}}),DeleteOne({"name": "Charlie"})
]result = collection.bulk_write(requests)
print("Inserted count:", result.inserted_count)
print("Matched count:", result.matched_count)
print("Modified count:", result.modified_count)
print("Deleted count:", result.deleted_count)

9. 事务

对于需要原子性操作的场景，可以使用 MongoDB 的事务功能。事务允许在多个操作中保持一致性

# 事务
from pymongo import MongoClient
from pymongo.errors import ConnectionFailureclient = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['mycollection']try:with client.start_session() as session:with session.start_transaction():collection.insert_one({"name": "Frank", "age": 33}, session=session)collection.update_one({"name": "Eve"}, {"$set": {"age": 30}}, session=session)
except ConnectionFailure as e:print("Transaction failed:", e)

参考资料

• PyMongo 官方文档
• MongoDB 官方文档