当前位置：首页 > 文章列表 > 文章 > python教程 > Python连接MongoDB实战教程

Python连接MongoDB实战教程

2025-08-14 19:42:51 0浏览收藏

想掌握Python连接MongoDB的精髓？本文为你深度解析pymongo库的使用技巧与优化策略。从基础的连接建立，到高级查询、数据聚合，再到错误处理与性能优化，一网打尽。通过本文，你将学会如何使用pymongo轻松实现MongoDB的增删改查操作，并掌握利用连接URI进行认证、配置连接池以提升高并发性能的方法。此外，文章还深入探讨了如何运用$gt、$in、$regex等操作符进行复杂查询，以及如何通过aggregate()方法构建强大的数据处理管道。更重要的是，本文还提供了实用的错误处理技巧和性能优化建议，助你打造稳定高效的Python MongoDB应用。

使用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

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

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()方法来使用聚合管道。

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

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

$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”这样的条件。