在“100天精通Python”系列的学习路线里第 40 天进入数据库操作阶段主角是 pymongo 和 MongoDB。MongoDB 是当前使用非常广泛的文档型 NoSQL 数据库它以 BSON 格式保存文档天然适合存储 JSON 风格的数据pymongo 则是 MongoDB 官方提供的 Python 驱动负责让 Python 程序与 MongoDB 服务端完成连接、读写和底层通信。对刚学完 Python 基础、第一次接触数据库联动开发的读者来说这一天的内容会把练习难度从“单机脚本”提升到“客户端与数据库配合”所以连接、写入、查询、更新、删除每一条链路都值得完整过一遍。这一天的内容会按一个可复现的顺序展开先说明 MongoDB 与关系型数据库的核心差异接着安装并启动 MongoDB再安装 pymongo然后建立连接逐步写完增删改查、排序分页、计数和简单聚合最后给出常见报错的排查路径和生产环境注意事项。学完这一篇你不仅能写出可运行的 pymongo 脚本还能在真正部署项目时知道哪些配置必须改、哪些写法不能照搬到生产环境。1. 先把 MongoDB 和 pymongo 的技术定位讲清楚1.1 MongoDB 是文档型数据库和 MySQL 的核心差异用一句话理解 MongoDB它是一种不强制表结构、以文档为单位存储数据的数据库。这里的“文档”不是 Word 文档而是一条 JSON 风格的记录MongoDB 内部会把这条记录序列化成 BSON 格式存储。和 MySQL 这类关系型数据库相比MongoDB 的核心差异体现在数据模型上。MySQL 要求先建库、建表、定义字段类型再插入数据MongoDB 则不需要提前定义集合结构插入第一条文档时才自动创建数据库和集合。MySQL 通过外键和 JOIN 把数据拆分到多张表MongoDB 更倾向于把一组相关数据放在同一个文档里减少跨表关联。对比维度MySQL关系型MongoDB文档型存储单位表、行、列集合、文档、字段表结构固定 schema字段类型严格动态 schema同集合内字段可以不一致查询方式SQL 语句查询表达式字典数据关系外键 JOIN嵌套文档 引用事务能力传统 ACID 事务成熟高版本也支持多文档事务但初学者先掌握文档模型更稳妥入门路径先学 SQL 语法和表设计先学 JSON、操作符和文档设计并不是说 MongoDB 可以完全替代 MySQL而是它们适用的场景不同。日志、用户画像、爬虫结果、配置类数据、频繁变动的 JSON 结构用 MongoDB 写起来很顺手强事务、强一致性、复杂报表关联关系型数据库仍然更稳。学习第 40 天时先接受它是一个“文档数据库”即可不需要急着下结论说谁更好。1.2 pymongo 是官方驱动负责 Python 和 MongoDB 之间的通信pymongo 不只是“操作 MongoDB 的 Python 工具”它本质上是 MongoDB 官方维护的 Python 客户端驱动。驱动要处理的底层工作比你想象得多建立和维护 TCP 连接管理连接池。把 Python 的 dict 转换为 BSON 格式发送给服务端。把服务端返回的 BSON 结果转换回 Python 的 dict。处理游标、心跳检测、副本集节点发现等底层机制。所以你在 pymongo 里写的大部分业务代码都是操作 dict 和查询条件字典并不需要手动拼接 JSON 字符串。这一点刚上手时可能感觉不到等后面接触非官方客户端或者自己实现协议时才能体会到驱动到底替你省了多少事。1.3 这一天的学习目标和前置要求学习这一天之前需要掌握 Python 基础语法包括 dict、list、循环、函数和异常处理。还需要能在命令行执行python和pip命令。数据库方面建议已经把 MongoDB 安装好并且会用mongosh查看数据再来练习 pymongo这样验证结果会容易很多。这一天的目标不是把 MongoDB 的所有特性讲完而是把程序里最常用的链路全部跑通连接、插入、查询、更新、删除、排序分页、计数和简单聚合。后面的章节全部围绕这几条链路展开。2. 环境准备MongoDB 服务端和 pymongo 都要装好2.1 MongoDB 服务端的安装、启动与状态检查pymongo 是客户端必须连接到一个正在运行的 MongoDB 服务端。不同操作系统安装 MongoDB 的方式差别很大官方安装文档会更准确这里重点说明安装完成之后如何启动和检查。在 Linux systemd 环境下安装完成后的典型操作是sudo systemctl start mongod sudo systemctl enable mongod sudo systemctl status mongod执行systemctl status mongod时预期能看到active (running)状态。Windows 下安装后MongoDB 通常注册为系统服务可以直接在服务管理器中确认运行状态macOS 使用 Homebrew 安装时可以通过brew services list查看。服务启动后使用mongosh验证服务端是否可连接mongosh --eval db.version()如果返回一个版本号字符串说明服务端已经正常监听默认端口 27017。注意不同操作系统、不同 MongoDB 版本的安装命令并不完全一样。上面命令用于说明安装完成后的启动与检查思路实际安装前要先确认自己的系统和目标版本。2.2 确认 Python 版本并安装 pymongopymongo 4.x 要求 Python 3.7 以上。先确认当前环境python --version pip --version安装 pymongopip install pymongo建议在虚拟环境中安装避免污染全局 Python 环境。如果已经进入虚拟环境直接用上面的命令即可。验证安装结果python -c import pymongo; print(pymongo.__version__)能输出版本号说明 pymongo 已经可以被 Python 正常导入。2.3 环境检查清单启动和安装完成后建议按下面的表格逐项检查一次避免后面写代码时把“环境问题”误判成“代码问题”。检查项检查命令预期结果MongoDB 服务sudo systemctl status mongodactive (running)服务端可连接mongosh --eval db.version()版本号字符串Python 版本python --version3.7 及以上pymongo 已安装python -c import pymongo; print(pymongo.__version__)版本号端口可连接运行第 3 章最小连接脚本无超时异常环境这一关过了后面才能把注意力放在 pymongo 的 API 和查询写法上。3. 建立连接并理解 Database、Collection 和文档3.1 MongoClient 连接串的构成pymongo 的连接入口是MongoClientfrom pymongo import MongoClient client MongoClient(mongodb://localhost:27017/)这个连接串由几部分组成连接串片段含义说明mongodb://协议头MongoDB 默认协议localhost主机地址本机环境使用27017端口MongoDB 默认监听端口/school数据库名可省略连接后再选择数据库user:pass用户名密码开启认证时才需要一个带认证的典型连接串是client MongoClient(mongodb://admin:123456localhost:27017/admin?authSourceadmin)这里authSourceadmin表示认证数据库是 admin而不是业务数据库。有一个排查时很重要的特性MongoClient创建时不会立刻发起显式的业务请求真正执行数据库操作时才建立连接。所以连接串写错了往往不是创建MongoClient时报错而是第一次执行insert_one、find等操作时才抛出异常。3.2 获取数据库和集合的方式连接成功后可以通过字典方式获取数据库和集合db client[school] students db[students]也可以使用属性方式db client.school students db.students两种方式等价。属性方式写起来快但当集合名与 pymongo 的方法名或属性名冲突时会有问题。比如某个集合恰好叫insert_onedb.insert_one会被解析成方法而不是集合。因此推荐使用client[school]、db[students]这种括号写法尤其当集合名包含特殊字符或与关键字冲突时。关键认知在 MongoDB 中数据库和集合都可以隐式创建。第一次向students集合写入文档时school数据库和students集合会自动出现不需要提前执行建库建表语句。3.3 插入第一条数据验证连接先跑一个最小插入脚本完整验证环境from pymongo import MongoClient client MongoClient(mongodb://localhost:27017/) db client[school] students db[students] doc { name: 张三, age: 20, class_name: Python 提高班, } result students.insert_one(doc) print(result.inserted_id) client.close()正常输出是一个ObjectId(...)例如ObjectId(6779f1a2b4c3d4e5f6a7b8c9)这个ObjectId是 MongoDB 自动生成的_id字段值。insert_one返回的是InsertOneResult对象inserted_id就是新文档的主键。此时切换到mongosh里可以看到同样的数据use school db.students.find().pretty()输出中会显示_id、name、age、class_name四个字段。到这一步说明 pymongo 与 MongoDB 的整个通信链路已经打通。4. 增删改查完整实战4.1 插入单个文档和批量插入单个文档插入使用insert_onedoc {name: 李四, age: 22, class_name: Python 基础班} result students.insert_one(doc) print(result.inserted_id)批量插入使用insert_many参数是一个文档列表docs [ {name: 王五, age: 21, class_name: Python 基础班}, {name: 赵六, age: 19, class_name: Python 提高班}, {name: 孙七, age: 23, class_name: 数据分析班}, ] result students.insert_many(docs) print(result.inserted_ids)insert_many返回InsertManyResultinserted_ids是一个ObjectId列表顺序与传入的文档顺序一致。插入时要注意如果没有显式提供_id字段MongoDB 会自动生成ObjectId。如果显式提供了_id那么同一个集合内不能重复否则会抛出DuplicateKeyError。4.2 查询单个文档和遍历查询结果查询单个文档使用find_one返回一个 dict 或Noneone students.find_one({name: 张三}) print(one)输出示例{_id: ObjectId(6779...), name: 张三, age: 20, class_name: Python 提高班}注意_id是ObjectId类型不是字符串。直接打印时看到ObjectId(...)是正常的。查询多个文档使用find返回的是一个 Cursor 对象cursor students.find() for s in cursor: print(s[name], s[age])Cursor 有两个容易踩坑的特点。第一它是惰性的遍历时才真正向服务端拉取数据第二它只能从头到尾迭代一次如果需要多次使用要先转换成列表all_students list(students.find()) print(len(all_students))4.3 条件查询比较运算符、逻辑运算符和字段判断pymongo 的查询条件就是 Python 字典操作符以$开头。最常用的条件写法如下操作符含义示例$eq等于{age: {$eq: 20}}$ne不等于{age: {$ne: 20}}$gt大于{age: {$gt: 20}}$gte大于等于{age: {$gte: 20}}$lt小于{age: {$lt: 30}}$lte小于等于{age: {$lte: 30}}$in在列表中{age: {$in: [19, 20]}}$nin不在列表中{age: {$nin: [19, 20]}}$and逻辑与{$and: [{age: {$gte: 18}}, {age: {$lt: 30}}]}$or逻辑或{$or: [{age: 19}, {age: 23}]}$exists字段是否存在{remark: {$exists: True}}$regex正则匹配{name: {$regex: ^张}}结合集合里的测试数据实际写法如下# 年龄大于等于 20 且小于 30可以直接把范围写在一个字段条件里 cursor students.find({age: {$gte: 20, $lt: 30}}) for s in cursor: print(s[name], s[age]) # 查询基础班或提高班的学生 cursor students.find({class_name: {$in: [Python 基础班, Python 提高班]}}) for s in cursor: print(s[name], s[class_name]) # 查询姓名以“张”开头的人 cursor students.find({name: {$regex: ^张}}) for s in cursor: print(s[name])多条件默认就是 AND 语义直接写多个字段即可cursor students.find({class_name: Python 提高班, age: {$gte: 20}})4.4 更新文档$set、$inc、$push 与 update_many更新使用update_one和update_many它们都接收两个字典第一个是过滤条件第二个是更新操作。result students.update_one( {name: 张三}, {$set: {age: 26}} ) print(result.matched_count) # 匹配到的文档数 print(result.modified_count) # 实际修改的文档数更新操作符最常用的是下面几个操作符作用示例$set修改字段值或新增字段{$set: {age: 25}}$inc数值加减{$inc: {score: 5}}$unset删除字段{$unset: {remark: }}$push向数组字段追加元素{$push: {tags: 优秀}}$pull从数组字段删除匹配元素{$pull: {tags: 优秀}}批量更新示例# 给基础班所有学生加 10 分 result students.update_many( {class_name: Python 基础班}, {$inc: {score: 10}} ) print(result.modified_count)一个必须区分清楚的点matched_count表示有多少条文档匹配了过滤条件modified_count表示实际发生了修改的条数。如果把某人的 age 从 26 改成 26matched_count是 1modified_count却是 0。排查更新不生效时这两个数字要分开看。4.5 删除文档按条件删除与清空集合删除使用delete_one和delete_manyresult students.delete_one({name: 李四}) print(result.deleted_count) result students.delete_many({class_name: Python 提高班}) print(result.deleted_count)清空整个集合可以使用空过滤条件result students.delete_many({}) print(result.deleted_count)delete_many({})会删除集合内所有文档但集合本身还存在。如果想把集合也一起删掉可以执行students.drop()学习环境随意删没问题生产环境一定要谨慎。delete_many的过滤条件写空字典时一次就会清空数据所以执行前必须确认条件是否正确。5. 排序、分页、计数与聚合5.1 sort、skip、limit 组合实现分页排序使用sort方法。按年龄升序cursor students.find().sort(age, 1) for s in cursor: print(s[name], s[age])sort的第二个参数1表示升序-1表示降序等价写法是from pymongo import ASCENDING, DESCENDING cursor students.find().sort(age, ASCENDING)多字段排序时传入一个由元组组成的列表cursor students.find().sort([(age, -1), (name, 1)])先按年龄降序年龄相同的人再按姓名升序。分页查询使用skip和limit组合page 2 page_size 3 cursor students.find().sort(age, -1).skip((page - 1) * page_size).limit(page_size) for s in cursor: print(s[name], s[age])这里skip跳过前面的数据limit限制返回数量。第 2 页每页 3 条就是跳过 3 条再取 3 条。当数据量很大时skip跳过的行数越多性能越差。生产环境做深分页时更推荐基于_id或排序字段的范围查询来取下一页而不是无限增大skip。5.2 count_documents 统计文档数量统计数量使用count_documentstotal students.count_documents({}) basic_count students.count_documents({class_name: Python 基础班}) print(total, basic_count)一个常见的兼容性坑旧版本 pymongo 里的count()方法从 4.0 开始已经移除继续调用会报AttributeError。遇到这个问题统一改成count_documents即可。5.3 aggregate 实现简单的分组统计聚合是 MongoDB 里功能最强的一部分第 40 天先掌握最简单的分组统计即可。比如按班级分组统计每个班级的平均分和人数pipeline [ { $group: { _id: $class_name, avg_score: {$avg: $score}, count: {$sum: 1} } }, {$sort: {avg_score: -1}} ] for item in students.aggregate(pipeline): print(item)输出示例{_id: Python 提高班, avg_score: 95.0, count: 1} {_id: Python 基础班, avg_score: 82.0, count: 2}aggregate接收的是一个列表列表中的每个字典代表一个聚合阶段数据会按顺序流经这些阶段。$group的_id指定分组字段$avg计算平均值$sum累加计数最后一个$sort对分组结果排序。聚合管道的学习曲线比普通查询陡一些但理解了这个“管道逐级处理数据”的思路后面看$lookup、$unwind这些操作符就会顺很多。6. 通过一个完整脚本走通全流程并验证6.1 完整示例脚本把前面的内容整理成一个完整脚本保存为demo_pymongo.pyfrom pymongo import MongoClient def main(): client MongoClient(mongodb://localhost:27017/) db client[school] students db[students] # 清空集合保证每次运行结果一致 students.delete_many({}) # 批量插入测试数据 students.insert_many([ {name: 张三, age: 20, class_name: Python 基础班, score: 88}, {name: 李四, age: 22, class_name: Python 提高班, score: 95}, {name: 王五, age: 21, class_name: Python 基础班, score: 76}, {name: 赵六, age: 19, class_name: 数据分析班, score: 69}, ]) print(总人数:, students.count_documents({})) print(按班级统计平均分:) pipeline [ {$group: {_id: $class_name, avg_score: {$avg: $score}}} ] for item in students.aggregate(pipeline): print(item) # 王五加 5 分 result students.update_one( {name: 王五}, {$inc: {score: 5}} ) print(更新匹配数:, result.matched_count, 实际修改数:, result.modified_count) print(按年龄升序展示:) for s in students.find().sort(age, 1): print(s[name], s[age], s[score]) client.close() if __name__ __main__: main()运行脚本python demo_pymongo.py正常输出大致如下总人数: 4 按班级统计平均分: {_id: Python 提高班, avg_score: 95.0} {_id: 数据分析班, avg_score: 69.0} {_id: Python 基础班, avg_score: 81.5} 更新匹配数: 1 实际修改数: 1 按年龄升序展示: 赵六 19 69 张三 20 88 王五 21 81 李四 22 95脚本里先执行delete_many({})清空集合是为了保证脚本可以重复运行且每次结果一致。这种写法只适合学习环境生产环境不要轻易这样清空数据。6.2 与 mongosh 交叉验证结果脚本运行后可以在mongosh中交叉验证数据use school db.students.find().sort({age: 1}).pretty() db.students.countDocuments()输出中的文档内容和脚本打印的结果应该一致。pymongo 与 mongosh 只是两个不同的客户端访问的是同一个服务端和同一份数据结果不可能互相矛盾。如果两边对不上优先怀疑是否连接了不同的数据库或不同的 MongoDB 实例。6.3 验证的关键点不要只验证脚本不报错还要核对数据本身的正确性inserted_id是否生成了ObjectId。find_one返回值是不是 dict找不到时是不是None。更新后matched_count与modified_count是否符合预期。删除后deleted_count是否正确。聚合结果的分组和平均值是否正确。脚本结束时client.close()是否执行。注意程序能启动、没有异常只代表语法和执行路径没问题不代表业务逻辑正确。数据库程序必须把写入、更新、删除后的实际数据再查出来核对一遍。7. 常见报错与排查链路7.1 ServerSelectionTimeoutError连接不上现象pymongo.errors.ServerSelectionTimeoutError: localhost:27017: [Errno 111] Connection refused可能原因MongoDB 服务没有启动。服务端口被占用或改成了其他端口。防火墙拦截了 27017 端口。检查顺序sudo systemctl status mongod sudo ss -tlnp | grep 27017如果服务未运行先启动服务如果端口不是 27017需要把连接串改成对应端口如果启用了防火墙确认是否放行了 MongoDB 端口。7.2 OperationFailure认证失败现象pymongo.errors.OperationFailure: Authentication failed.可能原因连接串里没有带用户名密码。用户名或密码错误。用户创建在 admin 库但连接时没有指定authSourceadmin。处理方式在连接串中显式指定认证库和账号。client MongoClient(mongodb://myUser:myPasslocalhost:27017/admin?authSourceadmin)开发环境学习时可以先不开认证但生产环境必须开启认证并且按最小权限给应用创建专用账号不要直接使用 root。7.3 用字符串查询 _id 导致查不到结果现象日期字符串能打印出来但find_one({_id: 6779f1a2b4c3d4e5f6a7b8c9})返回None。原因_id是ObjectId类型不是字符串。直接用字符串查询等于拿字符串去匹配ObjectId必然查不到。处理方式查询前先转换类型。from bson.objectid import ObjectId student students.find_one({_id: ObjectId(6779f1a2b4c3d4e5f6a7b8c9)}) print(student)7.4 更新后 modified_count 为 0现象update_one返回的matched_count是 1但modified_count是 0。原因$set设置的字段值和当前值相同MongoDB 认为没有实际变化所以不算修改。这不是 bug是 MongoDB 的更新语义。处理方式如果业务关心“匹配到但未变更”的情况用matched_count判断如果只关心是否改动了数据用modified_count。7.5 调用 count() 方法报 AttributeError现象AttributeError: Collection object has no attribute count原因pymongo 4.0 已经移除了count()方法。处理方式统一使用count_documents({})。常见报错可以整理成一张速查表问题现象常见原因检查方式处理建议ServerSelectionTimeoutError服务未启动、端口不对、防火墙拦截systemctl status mongod、检查端口启动服务修正连接串端口Authentication failed账号密码错误、认证库错误确认连接串用户、密码、authSource修正认证信息并最小权限授权按 _id 查不到数据用字符串匹配 ObjectId打印type(doc[_id])确认类型用ObjectId()转换后再查modified_count 为 0字段值没有实际变化打印更新前后字段值按业务使用 matched_count 判断count() AttributeErrorpymongo 4.0 移除旧方法查看 pymongo 版本改用count_documents
Python操作MongoDB：pymongo增删改查与聚合实战详解
在“100天精通Python”系列的学习路线里第 40 天进入数据库操作阶段主角是 pymongo 和 MongoDB。MongoDB 是当前使用非常广泛的文档型 NoSQL 数据库它以 BSON 格式保存文档天然适合存储 JSON 风格的数据pymongo 则是 MongoDB 官方提供的 Python 驱动负责让 Python 程序与 MongoDB 服务端完成连接、读写和底层通信。对刚学完 Python 基础、第一次接触数据库联动开发的读者来说这一天的内容会把练习难度从“单机脚本”提升到“客户端与数据库配合”所以连接、写入、查询、更新、删除每一条链路都值得完整过一遍。这一天的内容会按一个可复现的顺序展开先说明 MongoDB 与关系型数据库的核心差异接着安装并启动 MongoDB再安装 pymongo然后建立连接逐步写完增删改查、排序分页、计数和简单聚合最后给出常见报错的排查路径和生产环境注意事项。学完这一篇你不仅能写出可运行的 pymongo 脚本还能在真正部署项目时知道哪些配置必须改、哪些写法不能照搬到生产环境。1. 先把 MongoDB 和 pymongo 的技术定位讲清楚1.1 MongoDB 是文档型数据库和 MySQL 的核心差异用一句话理解 MongoDB它是一种不强制表结构、以文档为单位存储数据的数据库。这里的“文档”不是 Word 文档而是一条 JSON 风格的记录MongoDB 内部会把这条记录序列化成 BSON 格式存储。和 MySQL 这类关系型数据库相比MongoDB 的核心差异体现在数据模型上。MySQL 要求先建库、建表、定义字段类型再插入数据MongoDB 则不需要提前定义集合结构插入第一条文档时才自动创建数据库和集合。MySQL 通过外键和 JOIN 把数据拆分到多张表MongoDB 更倾向于把一组相关数据放在同一个文档里减少跨表关联。对比维度MySQL关系型MongoDB文档型存储单位表、行、列集合、文档、字段表结构固定 schema字段类型严格动态 schema同集合内字段可以不一致查询方式SQL 语句查询表达式字典数据关系外键 JOIN嵌套文档 引用事务能力传统 ACID 事务成熟高版本也支持多文档事务但初学者先掌握文档模型更稳妥入门路径先学 SQL 语法和表设计先学 JSON、操作符和文档设计并不是说 MongoDB 可以完全替代 MySQL而是它们适用的场景不同。日志、用户画像、爬虫结果、配置类数据、频繁变动的 JSON 结构用 MongoDB 写起来很顺手强事务、强一致性、复杂报表关联关系型数据库仍然更稳。学习第 40 天时先接受它是一个“文档数据库”即可不需要急着下结论说谁更好。1.2 pymongo 是官方驱动负责 Python 和 MongoDB 之间的通信pymongo 不只是“操作 MongoDB 的 Python 工具”它本质上是 MongoDB 官方维护的 Python 客户端驱动。驱动要处理的底层工作比你想象得多建立和维护 TCP 连接管理连接池。把 Python 的 dict 转换为 BSON 格式发送给服务端。把服务端返回的 BSON 结果转换回 Python 的 dict。处理游标、心跳检测、副本集节点发现等底层机制。所以你在 pymongo 里写的大部分业务代码都是操作 dict 和查询条件字典并不需要手动拼接 JSON 字符串。这一点刚上手时可能感觉不到等后面接触非官方客户端或者自己实现协议时才能体会到驱动到底替你省了多少事。1.3 这一天的学习目标和前置要求学习这一天之前需要掌握 Python 基础语法包括 dict、list、循环、函数和异常处理。还需要能在命令行执行python和pip命令。数据库方面建议已经把 MongoDB 安装好并且会用mongosh查看数据再来练习 pymongo这样验证结果会容易很多。这一天的目标不是把 MongoDB 的所有特性讲完而是把程序里最常用的链路全部跑通连接、插入、查询、更新、删除、排序分页、计数和简单聚合。后面的章节全部围绕这几条链路展开。2. 环境准备MongoDB 服务端和 pymongo 都要装好2.1 MongoDB 服务端的安装、启动与状态检查pymongo 是客户端必须连接到一个正在运行的 MongoDB 服务端。不同操作系统安装 MongoDB 的方式差别很大官方安装文档会更准确这里重点说明安装完成之后如何启动和检查。在 Linux systemd 环境下安装完成后的典型操作是sudo systemctl start mongod sudo systemctl enable mongod sudo systemctl status mongod执行systemctl status mongod时预期能看到active (running)状态。Windows 下安装后MongoDB 通常注册为系统服务可以直接在服务管理器中确认运行状态macOS 使用 Homebrew 安装时可以通过brew services list查看。服务启动后使用mongosh验证服务端是否可连接mongosh --eval db.version()如果返回一个版本号字符串说明服务端已经正常监听默认端口 27017。注意不同操作系统、不同 MongoDB 版本的安装命令并不完全一样。上面命令用于说明安装完成后的启动与检查思路实际安装前要先确认自己的系统和目标版本。2.2 确认 Python 版本并安装 pymongopymongo 4.x 要求 Python 3.7 以上。先确认当前环境python --version pip --version安装 pymongopip install pymongo建议在虚拟环境中安装避免污染全局 Python 环境。如果已经进入虚拟环境直接用上面的命令即可。验证安装结果python -c import pymongo; print(pymongo.__version__)能输出版本号说明 pymongo 已经可以被 Python 正常导入。2.3 环境检查清单启动和安装完成后建议按下面的表格逐项检查一次避免后面写代码时把“环境问题”误判成“代码问题”。检查项检查命令预期结果MongoDB 服务sudo systemctl status mongodactive (running)服务端可连接mongosh --eval db.version()版本号字符串Python 版本python --version3.7 及以上pymongo 已安装python -c import pymongo; print(pymongo.__version__)版本号端口可连接运行第 3 章最小连接脚本无超时异常环境这一关过了后面才能把注意力放在 pymongo 的 API 和查询写法上。3. 建立连接并理解 Database、Collection 和文档3.1 MongoClient 连接串的构成pymongo 的连接入口是MongoClientfrom pymongo import MongoClient client MongoClient(mongodb://localhost:27017/)这个连接串由几部分组成连接串片段含义说明mongodb://协议头MongoDB 默认协议localhost主机地址本机环境使用27017端口MongoDB 默认监听端口/school数据库名可省略连接后再选择数据库user:pass用户名密码开启认证时才需要一个带认证的典型连接串是client MongoClient(mongodb://admin:123456localhost:27017/admin?authSourceadmin)这里authSourceadmin表示认证数据库是 admin而不是业务数据库。有一个排查时很重要的特性MongoClient创建时不会立刻发起显式的业务请求真正执行数据库操作时才建立连接。所以连接串写错了往往不是创建MongoClient时报错而是第一次执行insert_one、find等操作时才抛出异常。3.2 获取数据库和集合的方式连接成功后可以通过字典方式获取数据库和集合db client[school] students db[students]也可以使用属性方式db client.school students db.students两种方式等价。属性方式写起来快但当集合名与 pymongo 的方法名或属性名冲突时会有问题。比如某个集合恰好叫insert_onedb.insert_one会被解析成方法而不是集合。因此推荐使用client[school]、db[students]这种括号写法尤其当集合名包含特殊字符或与关键字冲突时。关键认知在 MongoDB 中数据库和集合都可以隐式创建。第一次向students集合写入文档时school数据库和students集合会自动出现不需要提前执行建库建表语句。3.3 插入第一条数据验证连接先跑一个最小插入脚本完整验证环境from pymongo import MongoClient client MongoClient(mongodb://localhost:27017/) db client[school] students db[students] doc { name: 张三, age: 20, class_name: Python 提高班, } result students.insert_one(doc) print(result.inserted_id) client.close()正常输出是一个ObjectId(...)例如ObjectId(6779f1a2b4c3d4e5f6a7b8c9)这个ObjectId是 MongoDB 自动生成的_id字段值。insert_one返回的是InsertOneResult对象inserted_id就是新文档的主键。此时切换到mongosh里可以看到同样的数据use school db.students.find().pretty()输出中会显示_id、name、age、class_name四个字段。到这一步说明 pymongo 与 MongoDB 的整个通信链路已经打通。4. 增删改查完整实战4.1 插入单个文档和批量插入单个文档插入使用insert_onedoc {name: 李四, age: 22, class_name: Python 基础班} result students.insert_one(doc) print(result.inserted_id)批量插入使用insert_many参数是一个文档列表docs [ {name: 王五, age: 21, class_name: Python 基础班}, {name: 赵六, age: 19, class_name: Python 提高班}, {name: 孙七, age: 23, class_name: 数据分析班}, ] result students.insert_many(docs) print(result.inserted_ids)insert_many返回InsertManyResultinserted_ids是一个ObjectId列表顺序与传入的文档顺序一致。插入时要注意如果没有显式提供_id字段MongoDB 会自动生成ObjectId。如果显式提供了_id那么同一个集合内不能重复否则会抛出DuplicateKeyError。4.2 查询单个文档和遍历查询结果查询单个文档使用find_one返回一个 dict 或Noneone students.find_one({name: 张三}) print(one)输出示例{_id: ObjectId(6779...), name: 张三, age: 20, class_name: Python 提高班}注意_id是ObjectId类型不是字符串。直接打印时看到ObjectId(...)是正常的。查询多个文档使用find返回的是一个 Cursor 对象cursor students.find() for s in cursor: print(s[name], s[age])Cursor 有两个容易踩坑的特点。第一它是惰性的遍历时才真正向服务端拉取数据第二它只能从头到尾迭代一次如果需要多次使用要先转换成列表all_students list(students.find()) print(len(all_students))4.3 条件查询比较运算符、逻辑运算符和字段判断pymongo 的查询条件就是 Python 字典操作符以$开头。最常用的条件写法如下操作符含义示例$eq等于{age: {$eq: 20}}$ne不等于{age: {$ne: 20}}$gt大于{age: {$gt: 20}}$gte大于等于{age: {$gte: 20}}$lt小于{age: {$lt: 30}}$lte小于等于{age: {$lte: 30}}$in在列表中{age: {$in: [19, 20]}}$nin不在列表中{age: {$nin: [19, 20]}}$and逻辑与{$and: [{age: {$gte: 18}}, {age: {$lt: 30}}]}$or逻辑或{$or: [{age: 19}, {age: 23}]}$exists字段是否存在{remark: {$exists: True}}$regex正则匹配{name: {$regex: ^张}}结合集合里的测试数据实际写法如下# 年龄大于等于 20 且小于 30可以直接把范围写在一个字段条件里 cursor students.find({age: {$gte: 20, $lt: 30}}) for s in cursor: print(s[name], s[age]) # 查询基础班或提高班的学生 cursor students.find({class_name: {$in: [Python 基础班, Python 提高班]}}) for s in cursor: print(s[name], s[class_name]) # 查询姓名以“张”开头的人 cursor students.find({name: {$regex: ^张}}) for s in cursor: print(s[name])多条件默认就是 AND 语义直接写多个字段即可cursor students.find({class_name: Python 提高班, age: {$gte: 20}})4.4 更新文档$set、$inc、$push 与 update_many更新使用update_one和update_many它们都接收两个字典第一个是过滤条件第二个是更新操作。result students.update_one( {name: 张三}, {$set: {age: 26}} ) print(result.matched_count) # 匹配到的文档数 print(result.modified_count) # 实际修改的文档数更新操作符最常用的是下面几个操作符作用示例$set修改字段值或新增字段{$set: {age: 25}}$inc数值加减{$inc: {score: 5}}$unset删除字段{$unset: {remark: }}$push向数组字段追加元素{$push: {tags: 优秀}}$pull从数组字段删除匹配元素{$pull: {tags: 优秀}}批量更新示例# 给基础班所有学生加 10 分 result students.update_many( {class_name: Python 基础班}, {$inc: {score: 10}} ) print(result.modified_count)一个必须区分清楚的点matched_count表示有多少条文档匹配了过滤条件modified_count表示实际发生了修改的条数。如果把某人的 age 从 26 改成 26matched_count是 1modified_count却是 0。排查更新不生效时这两个数字要分开看。4.5 删除文档按条件删除与清空集合删除使用delete_one和delete_manyresult students.delete_one({name: 李四}) print(result.deleted_count) result students.delete_many({class_name: Python 提高班}) print(result.deleted_count)清空整个集合可以使用空过滤条件result students.delete_many({}) print(result.deleted_count)delete_many({})会删除集合内所有文档但集合本身还存在。如果想把集合也一起删掉可以执行students.drop()学习环境随意删没问题生产环境一定要谨慎。delete_many的过滤条件写空字典时一次就会清空数据所以执行前必须确认条件是否正确。5. 排序、分页、计数与聚合5.1 sort、skip、limit 组合实现分页排序使用sort方法。按年龄升序cursor students.find().sort(age, 1) for s in cursor: print(s[name], s[age])sort的第二个参数1表示升序-1表示降序等价写法是from pymongo import ASCENDING, DESCENDING cursor students.find().sort(age, ASCENDING)多字段排序时传入一个由元组组成的列表cursor students.find().sort([(age, -1), (name, 1)])先按年龄降序年龄相同的人再按姓名升序。分页查询使用skip和limit组合page 2 page_size 3 cursor students.find().sort(age, -1).skip((page - 1) * page_size).limit(page_size) for s in cursor: print(s[name], s[age])这里skip跳过前面的数据limit限制返回数量。第 2 页每页 3 条就是跳过 3 条再取 3 条。当数据量很大时skip跳过的行数越多性能越差。生产环境做深分页时更推荐基于_id或排序字段的范围查询来取下一页而不是无限增大skip。5.2 count_documents 统计文档数量统计数量使用count_documentstotal students.count_documents({}) basic_count students.count_documents({class_name: Python 基础班}) print(total, basic_count)一个常见的兼容性坑旧版本 pymongo 里的count()方法从 4.0 开始已经移除继续调用会报AttributeError。遇到这个问题统一改成count_documents即可。5.3 aggregate 实现简单的分组统计聚合是 MongoDB 里功能最强的一部分第 40 天先掌握最简单的分组统计即可。比如按班级分组统计每个班级的平均分和人数pipeline [ { $group: { _id: $class_name, avg_score: {$avg: $score}, count: {$sum: 1} } }, {$sort: {avg_score: -1}} ] for item in students.aggregate(pipeline): print(item)输出示例{_id: Python 提高班, avg_score: 95.0, count: 1} {_id: Python 基础班, avg_score: 82.0, count: 2}aggregate接收的是一个列表列表中的每个字典代表一个聚合阶段数据会按顺序流经这些阶段。$group的_id指定分组字段$avg计算平均值$sum累加计数最后一个$sort对分组结果排序。聚合管道的学习曲线比普通查询陡一些但理解了这个“管道逐级处理数据”的思路后面看$lookup、$unwind这些操作符就会顺很多。6. 通过一个完整脚本走通全流程并验证6.1 完整示例脚本把前面的内容整理成一个完整脚本保存为demo_pymongo.pyfrom pymongo import MongoClient def main(): client MongoClient(mongodb://localhost:27017/) db client[school] students db[students] # 清空集合保证每次运行结果一致 students.delete_many({}) # 批量插入测试数据 students.insert_many([ {name: 张三, age: 20, class_name: Python 基础班, score: 88}, {name: 李四, age: 22, class_name: Python 提高班, score: 95}, {name: 王五, age: 21, class_name: Python 基础班, score: 76}, {name: 赵六, age: 19, class_name: 数据分析班, score: 69}, ]) print(总人数:, students.count_documents({})) print(按班级统计平均分:) pipeline [ {$group: {_id: $class_name, avg_score: {$avg: $score}}} ] for item in students.aggregate(pipeline): print(item) # 王五加 5 分 result students.update_one( {name: 王五}, {$inc: {score: 5}} ) print(更新匹配数:, result.matched_count, 实际修改数:, result.modified_count) print(按年龄升序展示:) for s in students.find().sort(age, 1): print(s[name], s[age], s[score]) client.close() if __name__ __main__: main()运行脚本python demo_pymongo.py正常输出大致如下总人数: 4 按班级统计平均分: {_id: Python 提高班, avg_score: 95.0} {_id: 数据分析班, avg_score: 69.0} {_id: Python 基础班, avg_score: 81.5} 更新匹配数: 1 实际修改数: 1 按年龄升序展示: 赵六 19 69 张三 20 88 王五 21 81 李四 22 95脚本里先执行delete_many({})清空集合是为了保证脚本可以重复运行且每次结果一致。这种写法只适合学习环境生产环境不要轻易这样清空数据。6.2 与 mongosh 交叉验证结果脚本运行后可以在mongosh中交叉验证数据use school db.students.find().sort({age: 1}).pretty() db.students.countDocuments()输出中的文档内容和脚本打印的结果应该一致。pymongo 与 mongosh 只是两个不同的客户端访问的是同一个服务端和同一份数据结果不可能互相矛盾。如果两边对不上优先怀疑是否连接了不同的数据库或不同的 MongoDB 实例。6.3 验证的关键点不要只验证脚本不报错还要核对数据本身的正确性inserted_id是否生成了ObjectId。find_one返回值是不是 dict找不到时是不是None。更新后matched_count与modified_count是否符合预期。删除后deleted_count是否正确。聚合结果的分组和平均值是否正确。脚本结束时client.close()是否执行。注意程序能启动、没有异常只代表语法和执行路径没问题不代表业务逻辑正确。数据库程序必须把写入、更新、删除后的实际数据再查出来核对一遍。7. 常见报错与排查链路7.1 ServerSelectionTimeoutError连接不上现象pymongo.errors.ServerSelectionTimeoutError: localhost:27017: [Errno 111] Connection refused可能原因MongoDB 服务没有启动。服务端口被占用或改成了其他端口。防火墙拦截了 27017 端口。检查顺序sudo systemctl status mongod sudo ss -tlnp | grep 27017如果服务未运行先启动服务如果端口不是 27017需要把连接串改成对应端口如果启用了防火墙确认是否放行了 MongoDB 端口。7.2 OperationFailure认证失败现象pymongo.errors.OperationFailure: Authentication failed.可能原因连接串里没有带用户名密码。用户名或密码错误。用户创建在 admin 库但连接时没有指定authSourceadmin。处理方式在连接串中显式指定认证库和账号。client MongoClient(mongodb://myUser:myPasslocalhost:27017/admin?authSourceadmin)开发环境学习时可以先不开认证但生产环境必须开启认证并且按最小权限给应用创建专用账号不要直接使用 root。7.3 用字符串查询 _id 导致查不到结果现象日期字符串能打印出来但find_one({_id: 6779f1a2b4c3d4e5f6a7b8c9})返回None。原因_id是ObjectId类型不是字符串。直接用字符串查询等于拿字符串去匹配ObjectId必然查不到。处理方式查询前先转换类型。from bson.objectid import ObjectId student students.find_one({_id: ObjectId(6779f1a2b4c3d4e5f6a7b8c9)}) print(student)7.4 更新后 modified_count 为 0现象update_one返回的matched_count是 1但modified_count是 0。原因$set设置的字段值和当前值相同MongoDB 认为没有实际变化所以不算修改。这不是 bug是 MongoDB 的更新语义。处理方式如果业务关心“匹配到但未变更”的情况用matched_count判断如果只关心是否改动了数据用modified_count。7.5 调用 count() 方法报 AttributeError现象AttributeError: Collection object has no attribute count原因pymongo 4.0 已经移除了count()方法。处理方式统一使用count_documents({})。常见报错可以整理成一张速查表问题现象常见原因检查方式处理建议ServerSelectionTimeoutError服务未启动、端口不对、防火墙拦截systemctl status mongod、检查端口启动服务修正连接串端口Authentication failed账号密码错误、认证库错误确认连接串用户、密码、authSource修正认证信息并最小权限授权按 _id 查不到数据用字符串匹配 ObjectId打印type(doc[_id])确认类型用ObjectId()转换后再查modified_count 为 0字段值没有实际变化打印更新前后字段值按业务使用 matched_count 判断count() AttributeErrorpymongo 4.0 移除旧方法查看 pymongo 版本改用count_documents
