数据存储：一文掌握存储数据到MongoDB详解

关于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 MongoClient

def main():
    # 连接 MongoDB
    client = 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 ConnectionFailure

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

with 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支持。在实际应用中，结合最佳实践，可以构建高性能、可维护的数据存储解决方案。