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