python对mongodb的增删查改

news2025/1/7 15:02:39

python对mongodb的增删查改

1. 安装 pymongo
2. 连接 MongoDB
3. 创建（插入）文档
- 插入单个文档
- 插入多个文档
4. 查询文档
- 查询单个文档
- 查询多个文档
- 复杂查询
- 嵌套查询
- 分页条件查询（通用模版）
5. 更新文档
- 更新单个文档
- 更新多个文档
- 更新嵌套文档
6. 删除文档
- 删除单个文档
- 删除多个文档
7. 处理复杂的文档结构
- 插入带有数组的文档
- 查询嵌套数组中的元素
- 更新嵌套数组中的元素
8. 批量操作
9. 事务

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 = True
        num = num + 1
        cond["_id"] = {"$gt": item["_id"]}
        
        # TODO 处理业务逻辑
    if not dealing:
        break
    print("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}}  # 将年龄加 1

result = 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 ConnectionFailure

client = 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)