Python如何连接MongoDB？pymongo操作指南-Python教程-PHP中文网

使用pymongo连接mongodb时，认证可通过在连接uri中指定用户名、密码、认证数据库和机制（如scram-sha-1）来实现，推荐此方式以集中管理连接信息；2. 连接池由mongoclient默认管理，可通过maxpoolsize、minpoolsize、waitqueuetimeoutms等参数配置，以优化高并发下的性能和稳定性；3. 高级查询可使用$gt、$in、$regex等操作符结合sort()、limit()等链式方法实现复杂筛选；4. 聚合操作通过aggregate()方法构建包含$match、$group、$project等阶段的管道，支持强大的数据处理能力；5. 错误处理需捕获connectionfailure和operationfailure等异常以确保应用健壮性；6. 性能优化关键在于创建索引、使用批量操作、合理投影字段、重用mongoclient实例以及避免skip()分页性能问题，从而提升查询效率和系统吞吐量。

Python如何连接MongoDB？pymongo操作指南

使用Python连接MongoDB，

pymongo

登录后复制

库是当之无愧的首选工具。它提供了一套直观且功能强大的API，让你可以轻松地与MongoDB数据库进行交互，无论是数据的增删改查，还是更复杂的聚合操作，都能通过它来完成。

解决方案

要开始使用

pymongo

登录后复制

，你需要先安装它：

pip install pymongo

登录后复制

接着，最基本的连接和操作流程通常是这样的：

立即学习“Python免费学习笔记（深入）”；

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure

# 1. 建立连接
# 默认连接到 localhost:27017
# 如果MongoDB运行在其他地方，需要指定主机和端口
try:
    client = MongoClient('localhost', 27017)
    # 或者 client = MongoClient('mongodb://username:password@host:port/authSource?authMechanism=SCRAM-SHA-1')
    print("成功连接到MongoDB！")
except ConnectionFailure as e:
    print(f"连接MongoDB失败: {e}")
    exit() # 连接失败直接退出，后续操作无法进行

# 2. 选择数据库
# 如果数据库不存在，MongoDB会在你第一次插入数据时自动创建
db = client.mydatabase # 访问名为 'mydatabase' 的数据库

# 3. 选择集合 (相当于关系型数据库的表)
# 同样，如果集合不存在，会在第一次插入数据时自动创建
collection = db.mycollection # 访问名为 'mycollection' 的集合

# --- 常用操作 ---

# 插入单条文档 (document)
def insert_one_example():
    post = {"author": "Mike",
            "text": "My first blog post!",
            "tags": ["mongodb", "python", "pymongo"],
            "date": "2023-10-26"}
    result = collection.insert_one(post)
    print(f"插入单条文档ID: {result.inserted_id}")

# 插入多条文档
def insert_many_example():
    new_posts = [
        {"author": "John",
         "text": "Another post by John.",
         "tags": ["mongodb", "python"],
         "date": "2023-10-27"},
        {"author": "Jane",
         "text": "Jane's thoughts.",
         "date": "2023-10-28"}
    ]
    result = collection.insert_many(new_posts)
    print(f"插入多条文档ID: {result.inserted_ids}")

# 查询单条文档
def find_one_example():
    # 查询 author 为 "Mike" 的文档
    mike_post = collection.find_one({"author": "Mike"})
    print("查询到Mike的文档:", mike_post)

# 查询多条文档
def find_many_example():
    # 查询所有文档
    print("\n所有文档:")
    for post in collection.find():
        print(post)

    # 查询 tags 包含 "python" 的文档
    print("\n包含 'python' 标签的文档:")
    for post in collection.find({"tags": "python"}):
        print(post)

# 更新单条文档
def update_one_example():
    # 将 author 为 "Mike" 的文档的 text 字段更新
    result = collection.update_one(
        {"author": "Mike"},
        {"$set": {"text": "Updated text for Mike's post."}}
    )
    print(f"更新了 {result.modified_count} 条文档。")

# 更新多条文档
def update_many_example():
    # 将所有 text 包含 "post" 的文档添加一个新字段 "status": "processed"
    result = collection.update_many(
        {"text": {"$regex": "post"}},
        {"$set": {"status": "processed"}}
    )
    print(f"批量更新了 {result.modified_count} 条文档。")

# 删除单条文档
def delete_one_example():
    # 删除 author 为 "Jane" 的文档
    result = collection.delete_one({"author": "Jane"})
    print(f"删除了 {result.deleted_count} 条文档。")

# 删除多条文档
def delete_many_example():
    # 删除所有 status 为 "processed" 的文档
    result = collection.delete_many({"status": "processed"})
    print(f"批量删除了 {result.deleted_count} 条文档。")

# 清空集合 (删除所有文档)
def delete_all_example():
    result = collection.delete_many({})
    print(f"清空集合，删除了 {result.deleted_count} 条文档。")

# 运行示例
# insert_one_example()
# insert_many_example()
# find_one_example()
# find_many_example()
# update_one_example()
# update_many_example()
# delete_one_example()
# delete_many_example()
# delete_all_example()

# 4. 关闭连接 (通常在应用程序生命周期结束时才关闭)
# client.close()

登录后复制

pymongo连接MongoDB时，如何处理认证与连接池？

在生产环境中，处理好数据库的认证和连接池是至关重要的，这直接关系到应用程序的安全性和性能。

认证 (Authentication)

当MongoDB启用了认证功能后，你不能直接连接，需要提供用户名和密码。

pymongo

登录后复制

提供了几种方式来处理认证：

在连接URI中指定： 这是我个人比较推荐的方式，因为它把所有连接信息都集中在一起，清晰明了。

client = MongoClient('mongodb://username:password@host:port/authSource?authMechanism=SCRAM-SHA-1')
# authSource 是认证数据库的名称，通常是 'admin' 或存储用户信息的数据库
# authMechanism 是认证机制，常用的有 SCRAM-SHA-1 或 SCRAM-SHA-256

登录后复制

通过参数传递：

client = MongoClient(host='localhost', port=27017, username='myuser', password='mypassword', authSource='admin')

登录后复制

在连接后进行认证： 这种方式不太常见，也不太推荐，因为它在连接建立后才进行认证，可能会暴露未认证的连接时间。
```
client = MongoClient('localhost', 27017)
db = client.admin
db.authenticate('myuser', 'mypassword')
```
登录后复制
选择哪种方式取决于你的应用架构和安全策略，但核心思想是确保你的凭据以安全的方式传递。

连接池 (Connection Pool)

pymongo

登录后复制

默认就实现了连接池，这意味着你创建的

MongoClient

登录后复制

实例会维护一组到MongoDB服务器的开放连接。当你需要执行数据库操作时，

pymongo

登录后复制

会从池中获取一个可用的连接，操作完成后再将其归还。这避免了每次操作都重新建立TCP连接的开销，显著提升了性能。

你可以通过

MongoClient

登录后复制

的参数来配置连接池的行为：

```
maxPoolSize
```
登录后复制
: 连接池中允许的最大连接数。默认是100。如果你的应用并发量很高，可能需要适当调大。
```
minPoolSize
```
登录后复制
: 连接池中维护的最小连接数。默认是0。设置为一个非零值可以确保在低峰期也有一定数量的连接可用，避免首次请求的延迟。
```
maxIdleTimeMS
```
登录后复制
: 连接在池中空闲多久后会被关闭（毫秒）。默认没有限制。
```
waitQueueTimeoutMS
```
登录后复制
: 当连接池满时，客户端等待获取连接的最长时间（毫秒）。如果超时，会抛出
```
ConnectionFailure
```
登录后复制
。默认没有限制。

# 配置连接池，例如最大连接数20，最小连接数5，等待超时时间1秒
client = MongoClient('localhost', 27017, maxPoolSize=20, minPoolSize=5, waitQueueTimeoutMS=1000)

登录后复制

我个人觉得，连接池这东西，初学可能觉得麻烦，但真到了生产环境，它就是性能和稳定性的定海神针。正确配置连接池能有效避免“连接数过多”或“连接建立缓慢”这类问题，让你的应用在面对高并发时依然游刃有余。

在pymongo中进行高级查询与数据聚合有哪些技巧？

MongoDB的强大之处不仅在于其灵活的文档模型，还在于其丰富的查询能力，特别是聚合框架。

pymongo

登录后复制

完美地映射了这些功能。

高级查询 (Advanced Queries)

除了简单的键值对查询，你还可以使用操作符进行更复杂的筛选：

比较操作符：
```
$gt
```
登录后复制
(大于),
```
$lt
```
登录后复制
(小于),
```
$gte
```
登录后复制
(大于等于),
```
$lte
```
登录后复制
(小于等于),
```
$ne
```
登录后复制
(不等于),
```
$in
```
登录后复制
(在数组中),
```
$nin
```
登录后复制
(不在数组中)。
```
# 查询年龄大于25的用户
collection.find({"age": {"$gt": 25}})
# 查询标签包含 "python" 或 "mongodb" 的文档
collection.find({"tags": {"$in": ["python", "mongodb"]}})
```
登录后复制

逻辑操作符：

$and

登录后复制

$or

登录后复制

$not

登录后复制

$nor

登录后复制

。

# 查询年龄大于20且状态为 "active" 的用户
collection.find({"$and": [{"age": {"$gt": 20}}, {"status": "active"}]})
# 查询年龄小于18或状态为 "inactive" 的用户
collection.find({"$or": [{"age": {"$lt": 18}}, {"status": "inactive"}]})

登录后复制

元素操作符：

$exists

登录后复制

(字段是否存在),

$type

登录后复制

(字段类型)。

# 查询存在 email 字段的文档
collection.find({"email": {"$exists": True}})
# 查询 name 字段类型为字符串的文档
collection.find({"name": {"$type": "string"}})

登录后复制

数组操作符：

$all

登录后复制

(包含所有指定元素),

$size

登录后复制

(数组大小)。

# 查询 tags 数组同时包含 "mongodb" 和 "python" 的文档
collection.find({"tags": {"$all": ["mongodb", "python"]}})
# 查询 tags 数组大小为 2 的文档
collection.find({"tags": {"$size": 2}})

登录后复制

正则表达式查询：

$regex

登录后复制

。

# 查询 author 字段以 "J" 开头的文档
collection.find({"author": {"$regex": "^J"}})

登录后复制

链式操作：

limit()

登录后复制

sort()

登录后复制

skip()

登录后复制

projection()

登录后复制

。

# 查询所有文档，按日期降序排列，只取前10条，且只返回 author 和 text 字段
for doc in collection.find({}, {"author": 1, "text": 1, "_id": 0}).sort("date", -1).limit(10):
    print(doc)

登录后复制

数据聚合 (Aggregation Framework)

聚合框架是MongoDB处理复杂数据分析和转换的利器。它通过一系列“管道”（pipeline）阶段来处理文档流，每个阶段对文档进行转换，然后将结果传递给下一个阶段。

pymongo

登录后复制

通过

aggregate()

登录后复制

方法来使用聚合管道。

说实话，刚接触聚合管道的时候，我也有点懵，感觉像是在搭乐高积木，一步步把数据塑造成想要的样子。但一旦掌握了，那简直是数据分析的利器。

PPT.CN,PPTCN,PPT.CN是什么,PPT.CN官网,PPT.CN如何使用

一键操作，智能生成专业级PPT

查看详情

以下是一些常用的聚合阶段：

```
$match
```
登录后复制
: 过滤文档，类似于SQL的
```
WHERE
```
登录后复制
子句。
```
$group
```
登录后复制
: 对文档进行分组，并对每个组执行聚合操作（如求和、平均值）。
```
$project
```
登录后复制
: 重新塑造文档的结构，选择、重命名或添加字段。
```
$sort
```
登录后复制
: 对文档进行排序。
```
$limit
```
登录后复制
: 限制输出文档的数量。
```
$unwind
```
登录后复制
: 将数组字段的每个元素解构为单独的文档。

# 假设集合中有如下文档：
# {"_id": 1, "item": "A", "price": 10, "quantity": 2, "date": "2023-01-01"}
# {"_id": 2, "item": "B", "price": 15, "quantity": 1, "date": "2023-01-01"}
# {"_id": 3, "item": "A", "price": 10, "quantity": 3, "date": "2023-01-02"}

# 示例：按商品 (item) 分组，计算每个商品的销售总量和总金额
pipeline = [
    {"$match": {"date": {"$gte": "2023-01-01"}}}, # 筛选日期
    {"$group": {
        "_id": "$item", # 按 item 字段分组
        "totalQuantity": {"$sum": "$quantity"}, # 计算 quantity 的总和
        "totalAmount": {"$sum": {"$multiply": ["$price", "$quantity"]}} # 计算 price * quantity 的总和
    }},
    {"$sort": {"totalAmount": -1}}, # 按总金额降序
    {"$project": { # 重新组织输出字段
        "product": "$_id",
        "salesQuantity": "$totalQuantity",
        "salesAmount": "$totalAmount",
        "_id": 0 # 不显示 _id 字段
    }}
]

results = collection.aggregate(pipeline)
print("\n聚合查询结果:")
for doc in results:
    print(doc)

登录后复制

掌握聚合管道能够让你在Python中对MongoDB数据进行极其灵活和强大的分析，远超简单的CRUD操作。

Python操作MongoDB时，常见的错误处理与性能考量？

在实际开发中，除了实现功能，妥善处理错误和优化性能同样重要。

错误处理 (Error Handling)

pymongo

登录后复制

在遇到问题时会抛出异常，通常是

pymongo.errors

登录后复制

模块下的特定异常。捕获这些异常是构建健壮应用的关键。

ConnectionFailure

登录后复制

: 连接MongoDB服务器失败时抛出，例如服务器未运行、网络问题、端口错误等。

from pymongo.errors import ConnectionFailure

try:
    client = MongoClient('localhost', 27017, serverSelectionTimeoutMS=5000) # 设置连接超时
    client.admin.command('ping') # 尝试ping服务器确认连接
    print("MongoDB连接成功！")
except ConnectionFailure as e:
    print(f"MongoDB连接失败或服务器不可达: {e}")

登录后复制

OperationFailure

登录后复制

: 当MongoDB服务器拒绝操作（例如权限不足、语法错误、数据验证失败）时抛出。

from pymongo.errors import OperationFailure

try:
    # 尝试一个可能失败的操作，例如在没有权限的数据库上创建索引
    db.some_collection.create_index("non_existent_field", unique=True)
except OperationFailure as e:
    print(f"MongoDB操作失败: {e}")
    if "E11000 duplicate key error" in str(e):
        print("可能是唯一索引冲突。")
    elif "not authorized" in str(e):
        print("权限不足，请检查用户角色。")

登录后复制

其他Python内置异常： 比如
```
TypeError
```
登录后复制
（参数类型不匹配）、
```
ValueError
```
登录后复制
（参数值无效）等，也需要考虑。

一个好的实践是使用

try-except

登录后复制

块来包裹所有数据库操作，并根据不同的异常类型进行日志记录、用户提示或重试逻辑。

性能考量 (Performance Considerations)

索引 (Indexing)： 这是MongoDB性能优化的基石，没有之一。为经常查询的字段（特别是用于
```
find()
```
登录后复制
、
```
sort()
```
登录后复制
、
```
aggregate
```
登录后复制
的
```
$match
```
登录后复制
和
```
$sort
```
登录后复制
阶段）创建合适的索引，能极大提升查询速度。
```
# 为 'author' 字段创建升序索引
collection.create_index("author")
# 创建复合索引，先按 'date' 降序，再按 'author' 升序
collection.create_index([("date", -1), ("author", 1)])
# 确保索引创建一次即可，重复创建不会报错但无意义
```
登录后复制
我发现很多初学者，包括我自己当年，都容易忽略索引的重要性。数据量小的时候没感觉，一旦上去了，没索引的查询简直是噩梦。所以，建立合适的索引，真的是性能优化的重中之重。

批量操作 (Batch Operations)： 尽可能使用

insert_many()

登录后复制

update_many()

登录后复制

delete_many()

登录后复制

或

bulk_write()

登录后复制

来替代单条操作的循环。这能显著减少网络往返次数，提高吞吐量。

# 批量插入
documents = [{"name": f"doc_{i}"} for i in range(1000)]
collection.insert_many(documents)

# 批量更新 (使用 bulk_write 更灵活)
from pymongo import UpdateOne
requests = [UpdateOne({"name": f"doc_{i}"}, {"$set": {"status": "processed"}}) for i in range(500)]
collection.bulk_write(requests)

登录后复制

投影 (Projection)： 只查询你需要的字段，而不是整个文档。这减少了网络传输的数据量和客户端内存消耗。
```
# 只获取 author 和 text 字段，不获取 _id
for doc in collection.find({}, {"author": 1, "text": 1, "_id": 0}):
    print(doc)
```
登录后复制
```
MongoClient
```
登录后复制
实例的重用：
```
MongoClient
```
登录后复制
实例是线程安全的，并且内部管理着连接池。在应用程序的整个生命周期中，应该只创建一次
```
MongoClient
```
登录后复制
实例并重复使用它，而不是在每次操作时都创建新的实例。
合理使用
```
skip()
```
登录后复制
和
limit()
登录后复制
进行分页：对于大型数据集的分页，
```
skip()
```
登录后复制
在跳过大量文档时效率会降低。考虑使用基于游标（
```
_id
```
登录后复制
或时间戳）的分页方式，即“上次查询的最后一个文档的某个字段值大于X”这样的条件。