当前位置：首页 > news >正文

python操作mongodb

news 来源：原创 2025/9/9 12:00:31

1、安装包

pyMongo是MongoDB官方推荐的Python驱动程序，它提供了访问MongoDB数据库所需的接口。安装PyMongo非常简单，可以通过pip包管理工具来安装最新版本：

pip install pymongo

安装完成后，我们可以使用以下Python代码来检查是否成功安装了PyMongo：

import pymongoprint("pymongo version: ", pymongo.__version__)

2、mongo连接

数据库连接管理是开发过程中十分关键的一环。合理地管理连接可以提高程序的性能，避免不必要的资源浪费。

2.1、配置MongoDB连接参数

MongoDB的URI格式如下：mongodb://用户名：密码@服务器IP或域名：端口。通常MongoDB的默认端口是27017。以下是一个连接字符串的示例：

connection_string = "mongodb://username:password@localhost:27017/"
# mongodb://root:mongo@123@192.168.203.6:27017/

简化使用默认参数：

connection_string = "mongodb://localhost:27017/"

2.2、建立数据库连接

使用PyMongo建立连接，通常只需要导入pymongo库并使用MongoClient类即可：

from pymongo import MongoClientclient = MongoClient(connection_string)

MongoClient创建了一个MongoDB连接实例。默认情况下，MongoClient实例会自动连接到MongoDB服务器。这里 connection_string 是之前我们设置的连接字符串。

2.3、pymongo连接测试

网上大都上述方式，但连接时总是失败。遇到两个错误：

S1、pymongo.errors.InvalidURI: Username and password must be escaped according to RFC 3986, use urllib.parse.quote_plus()

解决方式：python连接Mongo问题：pymongo.errors.InvalidURI: Username and password must be escaped according to RFC... - 简书

S2、pymongo.errors.ServerSelectionTimeoutError: Could not reach any servers in [('localhost', 27017)]

解决方式：pymongo pymongo.errors.ServerSelectionTimeoutError: Could not reach any servers in [('localhost', 27017)] - 你说夕阳很美 - 博客园

# 最终测试通过代码
from urllib import parse
from pymongo import MongoClient# 转义用户名和密码
user = parse.quote_plus("用户名") # 根据RFC 3986转义
passwd = parse.quote_plus("密码") # 根据RFC 3986转义
connection_string = "mongodb://{0}:{1}@IP:PORT/?directConnection=true".format(user, passwd) # ?directConnection=true是关键client = MongoClient(connection_string)# 连接到mydatabase数据库
db = client["mydatabase"]
# 尝试执行一个简单的命令，例如查看所有集合
collections = db.list_collection_names()
print(collections)# 关闭客户端连接
client.close()

未报错，则运行成功！

3、数据库操作

3.1、创建数据库

如，创建的数据库 testdb：

db = client["mydatabase"] # 创建的数据库
# 或 db = client.get_database("mydatabase")

在 MongoDB 中，数据库只有在内容插入后才会创建，就是说，数据库创建后要创建集合(数据表)并插入一个文档(记录)，数据库才会真正创建。

3.2、判断指定的数据库是否存在

可以读取 MongoDB 中的所有数据库，并判断指定的数据库是否存在：

dblist = client.list_database_names() # 查看所有数据库名称
if "test" in dblist:print("数据库已存在！")

4、集合基本操作

4.1、集合创建

MongoDB 中的集合类似 SQL 的表，MongoDB 使用数据库对象来创建集合，实例如下：

db = client["mydatabase"] # 创建数据库"mydatabase"
collection = db["mycollection"] # 创建集合"mycollection"
# 或 collection = database.get_collection("mycollection")

在 MongoDB 中，集合只有在内容插入后才会创建，就是说，创建集合(数据表)后要再插入一个文档(记录)，集合才会真正创建。

4.2、判断指定的集合是否存在

可以读取 MongoDB 数据库中的所有集合，并判断指定的集合是否存在：

db = myclient["mydatabase"]
collist = db_test.list_collection_names() # 获取数据库中所有集合的名称
if "mycollection" in collist:  # 判断 mycollection 集合是否存在print("集合已存在！")

5、文档基础增删改查

5.1、插入文档

5.1.1、插入单条数据

使用insert_one()方法向集合中插入单条数据。

# 插入单条数据
doc = {"name": "John", "age": 30, "city": "New York"}
result = collection.insert_one(doc)# 打印插入的文档的ID
print(result.inserted_id)

5.1.2、插入多条数据

使用insert_many()方法一次性插入多条数据。

# 插入多条数据
docs = [{"name": "Amy", "age": 25, "city": "Los Angeles"},{"name": "Hannah", "age": 22, "city": "Chicago"},{"name": "Michael", "age": 27, "city": "Boston"}
]
result = collection.insert_many(docs)# 打印插入的文档的ID列表
print(result.inserted_ids)

5.2、查询文档

5.2.1、查询所有文档

使用find()方法可以查询集合中的所有文档。

# 查询所有文档
for idx, doc in enumerate(collection.find()):print(idx, doc)

5.2.2、查询单个文档

使用find_one()方法可以查询并返回集合中的第一个匹配文档。

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

5.3.3、带条件查询

通过传递查询条件来过滤返回的文档。

# 带条件查询
docs = collection.find({"age": {"$gt": 25}})
for idx, doc in enumerate(docs):print(idx, doc)

5.3.4、模糊查询

MongoDB支持正则表达式进行模糊查询。

# 模糊查询
import redocs = collection.find({"name": re.compile("^J")})
for idx, doc in enumerate(docs):print(idx, doc)

或者使用MongoDB的$regex操作符：

# 使用$regex进行模糊查询
docs = collection.find({"name": {"$regex": "^J"}})
for idx, doc in enumerate(docs):print(idx, doc)

5.3、更新文档

5.3.1、更新单个文档

使用update_one()方法可以更新集合中第一个匹配的文档。

# 更新单个文档
query = {"name": "John"}
newvalues = {"$set": {"age": 31}}
result = collection.update_one(query, newvalues)# 打印匹配和修改的文档数量
print(result.matched_count, result.modified_count)

5.3.2、更新多个文档

如果想要更新所有匹配的文档，可以使用update_many()方法。

# 更新多个文档
query = {"age": {"$gt": 25}}
newvalues = {"$set": {"status": "senior"}}
result = collection.update_many(query, newvalues)# 打印匹配和修改的文档数量
print(result.matched_count, result.modified_count)

5.4、删除文档

5.4.1、删除单个文档

使用delete_one()方法删除集合中第一个匹配的文档。

# 删除单个文档
query = {"name": "Amy"}
result = collection.delete_one(query)# 打印删除的文档数量
print(result.deleted_count)

5.4.2、删除多个文档

如果想要删除所有匹配的文档，可以使用delete_many()方法。

# 删除多个文档
query = {"status": "senior"}
result = collection.delete_many(query)# 打印删除的文档数量
print(result.deleted_count)

6、索引

MongoDB支持索引来提高查询性能。你可以为集合中的字段创建索引。

6.1、创建索引

collection.create_index([("name", pymongo.ASCENDING)], unique=True)

6.2、为多个字段创建复合索引

collection.create_index([("name", pymongo.ASCENDING), ("age", pymongo.DESCENDING)])

7、聚合管道

MongoDB的聚合管道是一种强大的数据处理工具，可以对集合中的文档进行复杂的数据处理。

# 使用聚合管道
pipeline = [{"$match": {"age": {"$gt": 25}}},{"$group": {"_id": "$city", "count": {"$sum": 1}}}
]
results = collection.aggregate(pipeline)for idx, result in enumerate(results):print(idx, result)

这个示例中，我们首先通过$match阶段筛选出年龄大于25的文档，然后通过$group阶段按城市分组，并计算每个城市的文档数量。

8、事务

从MongoDB 4.0开始，支持多文档事务。这允许你在多个集合和数据库上执行一系列操作，同时保持数据的一致性和完整性。

from pymongo import MongoClient# 连接到MongoDB
client = MongoClient("localhost", 27017)# 确保MongoDB在副本集模式下运行（事务需要副本集）
db = client["mydatabase"]# 开始一个会话
with client.start_session() as session:with session.start_transaction():# 在此执行事务操作collection1.insert_one(doc1, session=session)collection2.insert_one(doc2, session=session)# 注意：事务需要在副本集或分片集群上运行

9、实用技巧和最佳实践

9.1、使用连接池

在Web应用程序中，频繁地创建和销毁MongoDB连接是不高效的，建议使用连接池来重用连接。为了优化性能，PyMongo内部实现了连接池机制。这意味着当多个请求需要访问MongoDB时，它们可以从连接池中获取已存在的连接，而无需每次都建立新的连接。这在大量并发请求的场景下，可以显著提高性能和资源利用率。

连接池的具体细节和内部实现可能比较复杂，但是对于应用开发者而言，只需要正确地配置连接字符串和连接池大小，大多数场景下PyMongo都能够为我们提供稳定的数据库连接服务。

from pymongo import MongoClient# 创建一个连接池
client = MongoClient("localhost", 27017, maxPoolSize=100)# 使用client进行数据库操作
db = client["mydatabase"]
collection = db["mycollection"]
# 操作完成后，连接会自动返回到连接池中

9.2、监控和日志

对于生产环境，监控MongoDB的性能和日志是非常重要的。确保启用MongoDB的日志记录，并使用适当的工具来监控数据库的性能。

9.3、使用游标超时

当使用大量数据时，游标可能会占用大量资源。使用游标时，可以设置超时时间以避免潜在的资源泄露。

# 设置游标超时时间
cursor = collection.find().batch_size(10).add_option(pymongo.cursor.NoCursorTimeout)

注意：通常不推荐在生产环境中使用NoCursorTimeout，因为它可能导致游标无限期地保持打开状态。

9.4、安全性

确保MongoDB实例受到适当的保护，包括网络层面的访问控制、身份验证和授权。

9.5、性能和优化

索引优化：确保为查询中常用的字段创建索引，但也要避免过多索引，因为索引会占用额外的磁盘空间并可能影响写操作的性能。

查询优化：优化查询语句，减少不必要的数据检索，使用覆盖索引来减少磁盘I/O。

批量操作：在可能的情况下使用批量插入、更新和删除操作来提高性能。

监控和调优：定期监控MongoDB的性能指标，如查询响应时间、索引命中率、内存使用等，并根据需要进行调优。

9.6、使用GridFS存储大文件

MongoDB的GridFS是一个用于存储和检索大文件的规范。它可以将大文件分割成多个较小的块，并将这些块存储在MongoDB的集合中。GridFS提供了与MongoDB API类似的接口来操作文件。

from pymongo import MongoClient
from gridfs import GridFS# 连接到MongoDB
client = MongoClient("localhost", 27017)
db = client["mydatabase"]# 创建GridFS实例
fs = GridFS(db)# 上传文件
with open("large_file.dat", "rb") as f:file_id = fs.put(f, filename="large_file.dat")# 下载文件
with fs.get(file_id) as f:with open("downloaded_large_file.dat", "wb") as out:out.write(f.read())

9.7、异步操作

虽然PyMongo本身不提供直接的异步API，但你可以使用如motor这样的库来与MongoDB进行异步交互。motor是PyMongo的一个异步版本，它基于Python的asyncio库。

import asyncio
from motor.motor_asyncio import AsyncIOMongoClientasync def main():client = AsyncIOMongoClient("localhost", 27017)db = client["mydatabase"]collection = db["mycollection"]# 异步插入文档await collection.insert_one({"name": "John Doe", "age": 30})# 异步查询async for idx, doc in enumerate(collection.find({"age": 30})):print(idx, doc)# 关闭连接await client.close()运行异步主函数asyncio.run(main())

9.8、备份和恢复

定期备份MongoDB数据库是非常重要的，以防数据丢失或损坏。MongoDB提供了多种备份和恢复的方法，包括使用mongodump和mongorestore命令行工具，以及第三方备份解决方案。

参考文章：

1、pymongo官方API文档

MongoDB PyMongo 文档 - PyMongo 驱动程序 v4.12 - MongoDB Docs