Python3 模块精讲：pymongo（第三方）超详细教程 ——MongoDB 连接 + 全 CURD 实战

目录

1. 前言
2. pymongo 模块简介
3. 环境安装与版本适配
4. MongoDB 基础认知
5. pymongo 四种连接方式实战
6. 数据库与集合获取操作
7. 文档增删改查全套实战（带注释）
8. 条件查询、分页、排序、投影高级用法
9. 索引创建与性能优化
10. 连接池配置与高并发适配
11. 异常捕获与容错处理
12. 单例模式封装全局 MongoDB 工具类
13. 生产环境最佳实践
14. 常见报错与解决方案汇总
15. 全文总结

1. 前言

MongoDB 是目前最流行的非关系型文档数据库，以高性能、高可用、易扩展、 schema 灵活著称，广泛应用于爬虫数据存储、后端业务开发、日志存储、物联网、AI 非结构化数据存储等场景。

而pymongo是 MongoDB 官方为 Python3 提供的第三方驱动库，也是 Python 操作 MongoDB 的标准首选库。本文从安装→连接→库 / 集合操作→增删改查→高级查询→索引→连接池→工具类封装→生产最佳实践全流程讲解，每一段都附带可直接运行代码 + 逐行详细注释，全程干货无废话，适合收藏当工具书日常查阅。

2. pymongo 模块简介

2.1 核心作用

• 实现 Python3 程序与 MongoDB 服务的 TCP 通信
• 支持单节点、副本集、分片集群、云数据库（SSL/TLS）连接
• 提供完整的文档增、删、改、查、聚合、索引、事务等 API
• 内置连接池，天然支持高并发请求复用连接
• 兼容 MongoDB 4.x/ 5.x/ 6.x 全系列版本

2.2 核心对象关系

• MongoClient：MongoDB 客户端连接对象
• Database：数据库对象
• Collection：集合对象（相当于 MySQL 表）
• Document：文档对象（相当于 MySQL 行记录，Python 字典格式）

3. 环境安装与版本适配

3.1 pip 安装 pymongo

bash

运行

# 安装最新稳定版 pip install pymongo # 指定版本安装（企业常用固定版本） pip install pymongo==4.3.3 # 升级版本 pip install --upgrade pymongo

3.2 版本适配对照表

表格

3.3 验证是否安装成功

python

运行

# 导入模块并打印版本 import pymongo # 输出版本号，无报错即安装成功 print("pymongo 当前版本：", pymongo.__version__)

4. MongoDB 基础认知

1. 数据库 (Database)：对应 MySQL 的库
2. 集合 (Collection)：对应 MySQL 的数据表
3. 文档 (Document)：对应 MySQL 的一行记录，JSON / 字典格式
4. 字段：对应 MySQL 列
5. 默认端口：27017
6. 默认无认证：本地开发默认不需要账号密码

5. pymongo 四种连接方式实战

5.1 方式一：最简本地无认证连接（开发常用）

python

运行

# 导入核心客户端类 from pymongo import MongoClient # 创建客户端，不传参数默认连接 localhost:27017 client = MongoClient() # 手动指定主机和端口 # client = MongoClient(host="127.0.0.1", port=27017) # 校验连接是否正常，ping命令连通服务 client.admin.command("ping") print("✅ 最简无认证连接 MongoDB 成功") # 关闭连接（程序退出可不用手动关，框架会自动回收） # client.close()

5.2 方式二：URL 字符串连接（推荐统一配置）

python

运行

from pymongo import MongoClient # 标准URI连接格式：mongodb://host:port mongo_uri = "mongodb://127.0.0.1:27017" # 通过URI字符串创建客户端 client = MongoClient(mongo_uri) # 连通性测试 client.admin.command("ping") print("✅ URI 字符串连接成功")

5.3 方式三：带账号密码认证连接（生产必备）

前提：MongoDB 已开启身份认证，并创建业务用户。

python

运行

from pymongo import MongoClient from pymongo.errors import OperationFailure try: # 带账号、密码、认证库连接 client = MongoClient( host="127.0.0.1", port=27017, username="root", # 用户名 password="123456", # 密码 authSource="admin" # 认证数据库，默认admin ) # 验证连通与认证 client.admin.command("ping") print("✅ 账号密码认证连接成功") except OperationFailure: print("❌ 认证失败：用户名/密码/认证库错误") except Exception as e: print("❌ 连接异常：", e)

5.4 方式四：URI 带账号密码 + 认证库（生产最常用）

python

运行

from pymongo import MongoClient # 格式：mongodb://user:pwd@host:port/?authSource=admin uri = "mongodb://root:123456@127.0.0.1:27017/?authSource=admin" client = MongoClient(uri) client.admin.command("ping") print("✅ 带认证URI连接成功")

6. 数据库与集合获取操作

6.1 获取数据库对象两种写法

python

运行

from pymongo import MongoClient # 建立连接 client = MongoClient("mongodb://127.0.0.1:27017") # 写法1：下标方式获取数据库 db1 = client["test_db"] # 写法2：属性方式获取数据库 db2 = client.test_db # 打印数据库名称 print("数据库名称：", db1.name)

6.2 获取集合对象两种写法

python

运行

# 基于数据库获取集合 db = client["test_db"] # 写法1：下标方式 col1 = db["user_collection"] # 写法2：属性方式 col2 = db.user_collection print("集合名称：", col1.name)

重点：MongoDB无需手动建库、建集合，插入第一条数据自动创建库和集合。

7. 文档增删改查全套实战（带详细注释）

7.1 插入单条文档 insert_one

python

运行

from pymongo import MongoClient # 连接数据库 client = MongoClient("mongodb://127.0.0.1:27017") # 指定数据库 db = client["demo_db"] # 指定集合 col = db["user"] # 定义单条文档，Python字典对应MongoDB文档 user_doc = { "username": "张三", "age": 25, "gender": "男", "salary": 8000, "hobby": ["看书", "跑步"] } # 插入单条数据 result = col.insert_one(user_doc) # 获取自动生成的主键 _id print("✅ 插入成功，文档ID：", result.inserted_id)

7.2 插入多条文档 insert_many

python

运行

# 构造多条文档列表 user_list = [ {"username": "李四", "age": 28, "gender": "女", "salary": 9000}, {"username": "王五", "age": 30, "gender": "男", "salary": 12000}, {"username": "赵六", "age": 26, "gender": "女", "salary": 8500} ] # 批量插入 result = col.insert_many(user_list) # 打印所有插入的ID print("✅ 批量插入成功，ID列表：", result.inserted_ids)

7.3 查询单条数据 find_one

python

运行

# 查询匹配条件的第一条数据 # 条件字典：键为字段名，值为匹配值 res = col.find_one({"username": "张三"}) # 打印查询结果，字典格式 if res: print("✅ 查询到单条数据：", res) else: print("⚠️ 未查询到数据")

7.4 查询多条数据 find

python

运行

# 查询所有数据，返回游标对象 cursor = col.find() # 遍历游标取值 for item in cursor: print("逐条文档：", item) # 条件查询：年龄等于28 cursor2 = col.find({"age": 28}) for item in cursor2: print("年龄28用户：", item)

7.5 更新单条数据 update_one

python

运行

# 更新条件 cond = {"username": "张三"} # 更新操作 $set 只修改指定字段，不覆盖整条文档 update_data = {"$set": {"age": 26, "salary": 8800}} # 执行单条更新 result = col.update_one(cond, update_data) # 匹配条数、修改条数 print("匹配文档数：", result.matched_count) print("修改文档数：", result.modified_count)

7.6 更新多条数据 update_many

python

运行

# 条件：性别为男 cond = {"gender": "男"} # 统一涨薪1000 update_data = {"$set": {"salary": 13000}} result = col.update_many(cond, update_data) print("批量匹配数：", result.matched_count) print("批量修改数：", result.modified_count)

7.7 删除单条数据 delete_one

python

运行

# 删除条件 cond = {"username": "赵六"} # 执行单条删除 result = col.delete_one(cond) print("删除文档数量：", result.deleted_count)

7.8 删除多条数据 delete_many

python

运行

# 删除所有年龄大于30的数据 cond = {"age": {"$gt": 30}} result = col.delete_many(cond) print("批量删除数量：", result.deleted_count)

8. 条件查询、分页、排序、投影高级用法

8.1 比较运算符

$gt大于、$lt小于、$gte大于等于、$lte小于等于、$ne不等于

python

运行

# 查询年龄大于25小于30的用户 cond = { "age": { "$gt": 25, "$lt": 30 } } res = col.find(cond) for item in res: print("年龄区间用户：", item)

8.2 排序 sort

python

运行

# 按年龄升序：age 1升序 -1降序 res = col.find().sort("age", 1) for item in res: print("按年龄升序：", item)

8.3 分页 limit + skip

python

运行

# 跳过2条，取3条，实现分页 res = col.find().skip(2).limit(3) for item in res: print("分页数据：", item)

8.4 投影（只返回指定字段）

python

运行

# 只显示 username、age，隐藏 _id res = col.find({}, {"_id": 0, "username": 1, "age": 1}) for item in res: print("投影字段：", item)

9. 索引创建与性能优化

9.1 创建单个索引

python

运行

# 对age字段创建升序索引 col.create_index("age") print("✅ 普通索引创建完成")

9.2 创建唯一索引

python

运行

# username 唯一索引，不能重复 col.create_index("username", unique=True) print("✅ 唯一索引创建完成")

9.3 查看所有索引

python

运行

index_list = col.list_indexes() for idx in index_list: print("索引信息：", idx)

9.4 删除索引

python

运行

# 删除指定索引 col.drop_index("age_1") # 删除全部索引 # col.drop_indexes()

10. 连接池配置与高并发适配

pymongo 默认自带连接池，生产环境必须手动配置池参数，避免连接耗尽。

python

运行

from pymongo import MongoClient # 配置连接池参数 client = MongoClient( "mongodb://127.0.0.1:27017", maxPoolSize=50, # 最大连接数 minPoolSize=5, # 最小空闲连接 maxIdleTimeMS=30000, # 空闲连接超时毫秒 serverSelectionTimeoutMS=5000 # 服务选择超时 ) client.admin.command("ping") print("✅ 连接池配置生效，高并发就绪")

参数说明：

• maxPoolSize：并发峰值最大可用连接
• minPoolSize：常驻空闲连接，避免频繁创建
• serverSelectionTimeoutMS：超时时间，防止程序卡死

11. 异常捕获与容错处理

python

运行

from pymongo import MongoClient from pymongo.errors import ( ConnectionFailure, ServerSelectionTimeoutError, OperationFailure ) def get_mongo_client(): try: client = MongoClient( "mongodb://root:123456@127.0.0.1:27017/?authSource=admin", serverSelectionTimeoutMS=3000 ) client.admin.command("ping") return client except ServerSelectionTimeoutError: print("❌ 超时：MongoDB未启动或IP端口错误") except OperationFailure: print("❌ 认证失败：账号密码错误") except ConnectionFailure: print("❌ 网络连接失败") except Exception as e: print("❌ 未知异常：", e) return None

12. 单例模式封装全局 MongoDB 工具类（项目直接可用）

python

运行

from pymongo import MongoClient from pymongo.errors import ServerSelectionTimeoutError, OperationFailure class MongoUtil: # 私有类变量，保存单例客户端 _instance = None _client = None # 私有化构造方法 def __init__(self): # 数据库配置，实际项目放配置文件 self.uri = "mongodb://127.0.0.1:27017" self.db_name = "demo_db" # 单例入口 @classmethod def get_instance(cls): if not cls._instance: cls._instance = MongoUtil() cls._instance._init_client() return cls._instance # 初始化连接 def _init_client(self): try: self._client = MongoClient( self.uri, maxPoolSize=50, minPoolSize=5, serverSelectionTimeoutMS=5000 ) # 连通校验 self._client.admin.command("ping") print("✅ 全局MongoDB单例连接初始化成功") except Exception as e: print("❌ MongoDB连接初始化失败：", e) # 获取数据库对象 def get_db(self): return self._client[self.db_name] # 获取集合对象 def get_collection(self, col_name): return self.get_db()[col_name] # 使用示例 if __name__ == "__main__": # 获取工具类单例 mongo = MongoUtil.get_instance() # 获取user集合 col = mongo.get_collection("user") # 查询一条数据 res = col.find_one({"age": 25}) print("工具类查询结果：", res)

13. 生产环境最佳实践

1. 必须使用单例 MongoClient：全局只创建一个客户端，复用连接池
2. 账号密码绝不硬编码：放入配置文件 / 环境变量
3. 开启认证 + 内网访问：禁止公网裸奔端口
4. 合理配置连接池大小：根据业务并发调整 maxPoolSize
5. 关键字段建立索引：避免全表扫描拖垮性能
6. 禁止随意使用 delete_many 空条件：防止清空整表
7. 所有操作加异常捕获：防止数据库异常导致程序崩溃
8. 使用 $set 更新：避免覆盖整条文档

14. 常见报错与解决方案汇总

1. ServerSelectionTimeoutError

   • 原因：MongoDB 未启动、端口不通、防火墙拦截
   • 解决：启动服务、放行 27017 端口、检查 IP

2. Authentication failed

   • 原因：账号密码错误、authSource 不是 admin
   • 解决：核对账号密码，指定 authSource=admin

3. DuplicateKeyError

   • 原因：唯一索引字段插入重复值
   • 解决：校验数据唯一性或删除唯一索引

4. Connection pool closed

   • 原因：重复关闭客户端、连接被意外断开
   • 解决：使用单例，不手动频繁 close

15. 全文总结

本文完整讲解了pymongo 从安装、四种连接方式、库集合操作、增删改查、高级查询、分页排序投影、索引优化、连接池、异常处理、单例工具类封装全套知识点。