pymongo 是 Python 操作 MongoDB 数据库最常用的驱动库。今天这篇内容不是重新抄一遍文档而是把 pymongo 操作 MongoDB 的完整路径拆开安装环境、连接数据库、增删改查、条件查询、排序分页、索引、批量写入以及最后常见的报错排查一次讲到位。看到这个标题先说明白这套内容适合谁刚学完 Python 基础想接触数据库的读者写爬虫或数据处理脚本需要把结果落地到 MongoDB 的读者已经在用 MongoDB Compass 或 mongosh 操作数据但还没用 Python 驱动完成自动化的读者。先给一个实测后的判断不要一上来就研究索引和聚合那样会把入门变成劝退。MongoDB 的模型本身不复杂真正的学习顺序应该是先把服务跑起来用 mongosh 或图形工具确认能访问再用 pymongo 写一条文档进去最后才看查询、更新、索引这些工程细节。下面按这个顺序拆。1. 先搞清楚 pymongo 在项目里的定位避免盲目使用1.1 MongoDB 是文档型数据库pymongo 是 Python 侧的翻译官MongoDB 不是一个关系型数据库它保存的是 BSON 格式的文档也就是一种二进制化以后的 JSON 结构。你在 Python 里看到的是一个字典{name: 张三, age: 20, city: 上海}在 MongoDB 里它就是一个文档。多个文档组成集合多个集合组成数据库。这个层级关系和 MySQL 的数据库、表、行列结构不一样更像是一个可以灵活扩展的 JSON 存储。Python 程序本身不能直接和 MongoDB 服务端通信。pymongo 做了三件事把 Python 的 dict、list、str、int、datetime 等对象转换成 BSON发送查询、插入、更新、删除等命令给 MongoDB 服务端把服务端返回的结果再解析成 Python 对象。这就是驱动库的作用。你不需要自己去拼 JSON 字符串也不需要关心 socket 连接细节。1.2 什么时候该用 MongoDB什么时候不该硬上有一点要提前说明MongoDB 不是所有项目的默认选择更不是 MySQL 的平替。我比较推荐的使用场景爬虫数据存储。抓取下来的字段经常不固定今天多一个字段明天少一个字段MongoDB 不用改表结构。埋点日志和事件数据。数据量大写入频繁字段结构松散。快速原型开发。后端接口还在调整过程中MongoDB 不需要提前设计复杂表结构。需要水平扩展的数据场景。MongoDB 的分片机制比关系型数据库更容易往分布式方向走。不太适合的场景强事务和强一致性要求高。虽然新版 MongoDB 支持多文档事务但使用复杂度明显更高。中后台系统大量联表查询。MongoDB 不是不能关联查询而是频繁 $lookup 会让性能变差。对表结构、字段类型有严格规范的金融或财务类系统。关系型数据库的约束能力更强。我见过有人把用户订单全量存进 MongoDB结果后面每个字段都要单独更新还需要事务最后只能再迁回 MySQL。这不是 MongoDB 不行而是选型的时候没有把约束条件想清楚。1.3 学习 pymongo 前的心态准备如果你只是想把数据存进去、查出来pymongo 的学习量并不大。核心 API 只有几个MongoClient连接服务端insert_one / insert_many插入find_one / find查询update_one / update_many更新delete_one / delete_many删除先把这几个用熟再去碰索引和聚合。我在实际写代码时也很少在一个新增功能里同时用上全部 API更多时候是查询、更新、删除这三类反复出现。2. 环境准备让 MongoDB 服务和 Python 这边先握手成功2.1 MongoDB 服务端安装Windows 和 Debian/Ubuntu 两条路径pymongo 只是客户端真正干活的是 MongoDB 服务端。先确认你已经把服务跑起来。Windows 下通常有两种方式。一种是直接去 MongoDB 官网下载 MSI 安装包安装时勾选安装服务另一种是用包管理器例如通过 chocolatey 安装。安装完成后可以在服务列表里找到 MongoDB Server确认状态是正在运行。Debian/Ubuntu 下安装时有几个容易踩的坑。不同发行版自带的软件源版本差异比较大有的源里还是老版本直接用 apt 安装可能装到一个功能不完整的包。比较稳妥的做法是先确认系统版本再决定用系统源还是 MongoDB 官方源。# Debian/Ubuntu 先更新索引 sudo apt update sudo apt install -y mongodb-org安装完成后启动服务并检查状态sudo systemctl start mongod sudo systemctl status mongod如果看到 active (running)说明服务已经起来。如果启动失败优先看日志sudo journalctl -u mongod -n 50日志是定位问题的第一入口不要瞎猜。安装完成后验证端口是否在监听。默认端口是 27017Windows 下可以用netstat -ano | findstr 27017Linux 下可以用ss -lntp | grep 27017只要看到监听状态就说明服务端已经准备好。2.2 Python 侧安装 pymongo 并验证导入这个环节经常出现一个报错ModuleNotFoundError: No module named pymongo。原因很简单pymongo 不是 Python 标准库不安装直接用肯定报错。安装命令很简单pip install pymongo如果你同时用 VSCode 写代码先确认当前解释器是哪个环境。很多人装完以后还在原来的 Python 环境里运行自然导入失败。可以先在命令行里做一次验证python -c import pymongo; print(pymongo.__version__)如果能打印出版本号说明安装成功。如果仍然提示缺失检查一下 pip 安装到了哪个环境再用那个环境的 Python 执行上述命令。2.3 连接前的基础检查先别急着写代码我建议做一张检查清单按顺序确认检查项命令或方法正常结果异常处理服务是否启动systemctl status mongod / 任务管理器activerunning看日志启动失败先修配置文件端口是否监听netstat / ss27017 端口处于 LISTEN检查配置里的 bindIp 和 port是否需要认证连接串里有没有用户名密码无认证可直接连使用 authSource 指定认证库Python 解释器环境python --version与 pip 环境一致切换解释器或重装 pymongo这里有一个安全提醒如果 MongoDB 仅在本机开发默认只监听 127.0.0.1 是最安全的。不要为了图方便把 bindIp 改成 0.0.0.0又关闭认证直接把数据库暴露到局域网或公网。生产环境一定要开启访问控制。3. 连接 MongoDB理解库、集合、文档三层结构再写代码3.1 最小连接代码先确认服务能通环境准备好以后第一步不是写增删改查而是写一个最小连接脚本确认服务端能响应。from pymongo import MongoClient client MongoClient(mongodb://localhost:27017/) # 通过 ping 命令验证连接 try: client.admin.command(ping) print(连接正常) except Exception as e: print(连接失败:, e)这里有一个很多人不知道的细节MongoClient 创建的时候并不会立刻发起网络连接。真正建立连接发生在执行第一条命令时比如上面的 ping。所以如果你在创建 client 之后没有执行任何操作程序不会报错但这不代表连接一定成功。本地开发时我一般会设置一个较短的 serverSelectionTimeoutMS避免默认 30 秒超时导致排查时间太长client MongoClient( mongodb://localhost:27017/, serverSelectionTimeoutMS2000 )这样如果服务没启动2 秒内就会抛出 ServerSelectionTimeoutError提示找不到合适的服务器节点。3.2 库、集合、文档的三层结构MongoDB 的逻辑结构是三层的数据库Database集合Collection文档Document在 pymongo 里获取数据库和集合对象非常直接db client[testdb] collection db[users]也可以使用属性方式db client.testdb collection db.users两种写法等价。我个人更习惯用字典方式因为数据库名和集合名是动态传入时更清晰。需要特别注意的是MongoDB 有懒创建机制。也就是说你不需要提前在服务端建库、建集合。当你第一次向某个集合插入文档时数据库和集合会自动创建。collection.insert_one({name: 测试, age: 1})执行完以后testdb 库和 users 集合都会出现。如果你查询一个不存在的集合find_one 会返回 None不会报错。想确认当前有哪些数据库和集合可以这样print(client.list_database_names()) print(db.list_collection_names())3.3 连接参数怎么选默认和进阶怎么取舍MongoDB 连接串常用参数不多先记住这几个就够用参数作用常见设置mongodb://localhost:27017本地默认端口连接开发环境最常用serverSelectionTimeoutMS服务节点选择超时时间本地调试设置 2000~5000connectTimeoutMSTCP 连接超时时间默认即可socketTimeoutMSsocket 读写超时时间大批量操作可适当调大authSource认证库通常是 admin开启认证时需要指定replicaSet副本集名称连接副本集时使用如果 MongoDB 没有开启认证连接串写成这样就行client MongoClient(mongodb://localhost:27017/)如果开启了账号密码连接串需要带用户信息client MongoClient(mongodb://user:passwordlocalhost:27017/admin)这里的 /admin 是认证库不一定是业务库。很多新手在这里出错用户名密码明明是对的但认证一直失败原因就是认证库写错了。4. 增删改查实战单条插入到批量删除一次跑通4.1 插入insert_one 和 insert_many进入正文的核心部分。先从插入开始。插入一条文档result collection.insert_one( {name: 张三, age: 20, city: 上海} ) print(result.inserted_id)inserted_id 是这条文档的 _id 字段的值。MongoDB 默认会给每条文档生成一个 ObjectId它是全局唯一的。要注意inserted_id 的类型不是字符串打印出来像这样642f8f1b9a3f7c2d1e4f2a1b如果需要把 _id 记录到日志或前端返回建议转成字符串str(result.inserted_id)插入多条文档时用 insert_manydocs [ {name: 李四, age: 22, city: 北京}, {name: 王五, age: 25, city: 广州}, {name: 赵六, age: 28, city: 上海}, ] result collection.insert_many(docs) print(result.inserted_ids)insert_many 返回的是一个列表里面包含每条文档的 _id。这里我建议一个实测顺序先插入三条以内的小数据确认返回结果和 _id 都正常再考虑批量插入大量数据。不要一上来就塞几万条后面排查问题会很麻烦。4.2 查询find_one 和 find 游标插入以后先查询验证。doc collection.find_one({name: 张三}) print(doc)find_one 返回的是一个字典如果找不到就返回 None。查询多条时用 findcursor collection.find() for doc in cursor: print(doc)find 返回的并不是一个 list而是一个 Cursor 游标对象。它可以在遍历时才从服务端取数据。有一个常见误区直接打印 cursor 会看到一个 Cursor 对象而不是数据列表。很多人在这里以为自己查询失败了其实只是没有遍历。如果确实想一次性拿到列表可以docs list(collection.find())但是要注意数据量很大的时候list 会占大量内存。我的建议是小数据量测试可以用 list生产环境尽量用游标遍历或分批处理。统计文档数量用的是 count_documentscount collection.count_documents({city: 上海}) print(count)这个不是 count()新版 pymongo 已经弃用旧写法建议直接用 count_documents。4.3 更新update_one 和 update_many过滤条件与修改器要分开更新是最容易写错的部分。核心是理解 filter 和 update 是两个独立的参数。result collection.update_one( {name: 张三}, {$set: {age: 21}} ) print(result.matched_count) print(result.modified_count)filter 是筛选条件$set 是修改器意思是将匹配到的文档的 age 字段更新为 21。常见修改器有$set设置字段值字段不存在会新增$inc数字自增$push向数组字段追加元素$pull从数组字段移除元素示例collection.update_one( {name: 张三}, {$inc: {age: 1}} ) collection.update_one( {name: 张三}, {$push: {tags: python}} )最容易踩的坑是只传了新文档忘了加修改器# 错误示范会把整个文档替换成只有 age 字段的新文档 collection.update_one({name: 张三}, {age: 21})这个操作会删除原来文档里的 name、city 等其他字段只保留 age。后果非常严重。更新多条用 update_manyresult collection.update_many( {city: 上海}, {$set: {region: 华东}} ) print(result.modified_count)还有个 upsert 参数表示如果条件匹配不到就插入一条新文档collection.update_one( {name: 钱七}, {$set: {age: 30, city: 深圳}}, upsertTrue )如果存在 name 为 钱七 的文档就更新不存在就插入。这个参数在处理同步场景时很好用。4.4 删除delete_one 和 delete_many先看影响条数再删删除和更新一样先写过滤条件。result collection.delete_one({name: 张三}) print(result.deleted_count)delete_one 只删除匹配的第一条。如果条件匹配多条只删一条。如果想清空某个城市的全部文档result collection.delete_many({city: 上海}) print(result.deleted_count)我很建议在任何删除操作之前先做一次查询确认匹配范围matches list(collection.find({city: 上海})) print(len(matches))确认这个数量就是你预期要删除的量再执行删除。否则一条错误的过滤条件可能把整个集合清空。5. 查询排序、分页和聚合的实用写法5.1 比较条件和逻辑组合$gt、$lt、$in、$or开发中查询条件和简单等值查询完全不同。最常见的需求是范围查询。例如查年龄在 20 到 30 之间的用户cursor collection.find( {age: {$gte: 20, $lt: 30}} )查城市在指定列表里的用户cursor collection.find( {city: {$in: [上海, 北京]}} )多个字段的条件默认是 AND 关系cursor collection.find( {city: 上海, age: {$gte: 20}} )如果需要 OR使用 $orcursor collection.find( { $or: [ {city: 上海}, {age: {$lt: 25}} ] } )多个运算符组合时建议先在 MongoDB Compass 或 mongosh 里验证一次确认结果符合预期再搬进 Python 代码。这样可以隔离问题先确认数据库本身能查出数据再确认 pymongo 写法正确。5.2 排序和分页sort、skip、limit 与更稳的游标分页排序用 sort 方法。import pymongo cursor collection.find().sort(age, pymongo.ASCENDING)也可以使用字符串写法cursor collection.find().sort(age, 1) # 1 升序-1 降序多个字段排序cursor collection.find().sort( [(city, 1), (age, -1)] )分页最朴素的方式page 1 page_size 10 cursor collection.find().sort(age, -1).skip((page - 1) * page_size).limit(page_size)skip limit 在数据量小时没有问题但数据量大了以后效率很差。因为 MongoDB 需要跳过大量文档才能返回目标位置。更好的方式是游标分页利用 _id 或排序字段作为游标last_id None page_size 10 # 第一页 cursor collection.find().sort(_id, 1).limit(page_size) docs list(cursor) last_id docs[-1][_id] if docs else None # 下一页 cursor collection.find( {_id: {$gt: last_id}} ).sort(_id, 1).limit(page_size)这种基于游标的分页方式在数据量大时更稳定也不会出现翻页时新增数据导致的数据重复或遗漏。5.3 正则查询模糊匹配但要注意索引效率模糊查询是常见需求MongoDB 支持用 $regex 实现。cursor collection.find( {name: {$regex: ^张}} )这里 ^张 表示以“张”开头。还支持不区分大小写cursor collection.find( {name: {$regex: ^zhang, $options: i}} )需要注意两点第一正则查询如果字段没有索引会做集合扫描数据量大时非常慢。第二中文前缀匹配要结合实际情况判断不能指望所有正则查询都走索引。我的经验是模糊查询只适合小数据量或后台管理场景真正的高频查询尽量用精确匹配或前缀索引。5.4 聚合 pipeline先跑通一个分组统计聚合是 MongoDB 里能力很强但入门容易懵的部分。建议先理解 pipeline 的概念数据像水流一样经过多个阶段依次处理。一个简单的分组统计例子按城市分组统计每个城市的人数再排序。pipeline [ {$match: {age: {$gte: 20}}}, {$group: {_id: $city, count: {$sum: 1}}}, {$sort: {count: -1}} ] for doc in collection.aggregate(pipeline): print(doc)解释一下每个阶段$match先过滤出满足条件的文档$group按 city 字段分组$sum 计数$sort按 count 降序再复杂一点的统计平均年龄pipeline [ {$group: { _id: $city, avg_age: {$avg: $age}, count: {$sum: 1} }} ]聚合 pipeline 可以叠加很多操作但是不建议一上来就写超长 pipeline。先写一个阶段验证结果再叠加下一个阶段。这样如果结果不对能快速定位是哪个阶段的问题。6. 索引、批量写入和连接复用生产化必过的三关6.1 创建索引为什么能改变查询速度没有索引MongoDB 查询会扫描集合里的每一个文档数据量大了以后速度急剧下降。创建索引以后数据库会维护一个额外的索引结构查询时直接定位到匹配位置。创建单字段索引collection.create_index(age)创建唯一索引collection.create_index(email, uniqueTrue)创建复合索引collection.create_index([(city, 1), (age, -1)])查看已有索引for index in collection.list_indexes(): print(index)我建议先按查询条件决定索引不要把所有字段都加索引。索引越多写入时维护成本越高磁盘占用也越大。最合理的做法是先记录高频查询条件再看哪些字段组合需要建立索引。注意唯一索引创建后如果插入重复值会抛出 DuplicateKeyError。这个报错在批量导入时经常出现不要把它当成普通网络错误。6.2 批量写入减少网络往返而不是 for 循环 insert_one很多新手在导入数据时会写这样的循环for doc in docs: collection.insert_one(doc)每条 insert_one 都是一次网络请求。数据量到几千条时速度会明显变慢。更好的做法是减少请求次数。如果只需要批量插入用 insert_manycollection.insert_many(docs)如果同时有插入、更新、删除用 bulk_writefrom pymongo import InsertOne, UpdateOne, DeleteOne requests [ InsertOne({name: 赵六, age: 30, city: 成都}), UpdateOne( {name: 钱七}, {$set: {age: 31}}, upsertTrue ), DeleteOne({name: 王五}), ] collection.bulk_write(requests)bulk_write 返回结果里有 inserted_count、matched_count、modified_count、deleted_count可以用于统计。还有一个参数叫 ordered。默认是 True表示按顺序执行遇到一条失败就停止如果设为 False会跳过错误继续执行适合清理一批脏数据时使用。collection.bulk_write(requests, orderedFalse)但是要小心orderedFalse 时失败操作会被忽略你需要在返回结果的 bulk_api_errors 里查看异常。6.3 连接复用不要在每次请求里创建 MongoClient这个问题在 Web 开发中很常见。错误示范def get_user(user_id): client MongoClient(mongodb://localhost:27017/) db client[testdb] collection db[users] return collection.find_one({_id: user_id})每次调用函数都创建新的连接连接来不及复用很快就会把连接池打满可能出现 Too many open files 或连接超时。正确做法在程序启动时创建一个全局 MongoClient后续一直复用。from pymongo import MongoClient client MongoClient( mongodb://localhost:27017/, maxPoolSize20, minPoolSize2 ) db client[testdb] collection db[users]MongoClient 内部自带连接池是线程安全的在多线程服务里可以直接复用。FastAPI、Django、Flask 这类框架应该把 client 的初始化放在应用启动阶段而不是放在每个请求处理函数里。连接串也建议放到环境变量或配置文件中不要硬编码在代码里。6.4 大批量导入时的资源和编码问题大批量导入时最容易遇到的问题不是语法而是资源消耗。首当其冲是内存。如果你一次性把一个很大的 JSON 文件用 json.load 读进 Python再整个传给 insert_many内存占用会非常大。更稳的做法是用生成器按批读取每批插入 1000 条。def docs_generator(file_path, batch_size1000): with open(file_path, r, encodingutf-8) as f: batch [] for line in f: batch.append(json.loads(line)) if len(batch) batch_size: yield batch batch [] if batch: yield batch for batch in docs_generator(data.jsonl): collection.insert_many(batch)编码问题也很常见。从 Windows 导出的 UTF-8 文件可能带 BOM直接用 json.load 会报错。可以用如下方式处理with open(file_path, r, encodingutf-8-sig) as f: ...如果导入过程中写入超时不一定是网络问题可能是单批数据量太大导致单次操作时间超过 socketTimeoutMS。可以把 batch_size 调小或者适当调大 socketTimeoutMS。7. 常见报错和排查顺序遇到问题按这个链路走7.1 连接不上Connection refused 和时间超时现象很明显程序抛出 ConnectionFailure 或 ServerSelectionTimeoutError。排查顺序先确认 MongoDB 服务是否启动。再确认端口是否正确默认端口 27017。然后看连接串里的 IP 和认证信息。最后确认防火墙是否拦截了端口。示例# 检查端口 netstat -ano | findstr 27017 # 从 Python 里测试网络 python -c import socket; print(socket.create_connection((localhost, 27017), timeout3))如果 socket.create_connection 成功说明网络可达问题可能在连接参数或认证。7.2 认证失败Authentication failed认证失败通常是三个原因用户名或密码错误认证库指定错误用户权限不足检查连接串里的路径部分client MongoClient(mongodb://user:passlocalhost:27017/admin)/ admin 表示使用 admin 库进行认证。如果用户是在其他业务库下创建的要改成对应库。7.3 InvalidDocumentPython 类型无法写入 MongoDBMongoDB 的 BSON 支持的类型是有限的。比较常见的写入报错是 InvalidDocument例如试图写入一个 Python set。# 会报错 collection.insert_one({tags: {python, mongodb}})set 在 Python 里很常用但 BSON 没有 set 类型。需要改成 listcollection.insert_one({tags: [python, mongodb]})datetime 是可以写入的但建议统一存 UTC 时间避免不同时区数据混乱。Python 类型是否可写入 BSON说明str是最常见的类型int是注意超出 64 位范围会报错float是正常支持bool是list是对应数组dict是对应嵌套文档set否需要先转成 listbytes是对应 BSON Binarydatetime是建议存 UTC 时间None是对应 null7.4 OperationFailure字段名和排序问题MongoDB 对字段名有保留限制字段名不能包含英文点号 . 或以美元符号 $ 开头。如果你从外部系统导入大量 JSON字段名里混进了这些字符写入时会报 OperationFailure。解决办法是导入前先做字段名清洗把 . 替换成下划线或者给字段名加一层包裹。排序时报错往往是因为排序字段不存在或者字段类型不一致。MongoDB 对不同类型字段排序有自己的顺序比如数字和字符串混在一个字段里时排序结果可能不符合直觉。对这种数据写入前就要统一字段类型。7.5 我的排查顺序遇到问题我一般按这个链路走看具体报错信息判断是连接层、认证层、语法层还是类型层。用 mongosh 或 MongoDB Compass 直接执行同样的操作确认是数据库本身的问题还是 pymongo 写法的问题。最小化复现把代码缩减到十几行去掉业务逻辑保留一个 insert 或 find。检查资源占用服务端 CPU、内存、磁盘以及客户端有没有频繁创建连接。检查数据本身字段类型、编码、重复键、字段名合法性。如果还排查不出来就打开 MongoDB 的日志看服务端在报错时刻记录了什么。服务端日志比客户端报错信息更接近根因。最后留一个经验我自己在项目里接入 pymongo 时最喜欢的流程是先用 mongosh 把数据结构验证一遍再写 pymongo 代码。这样做的好处是数据库语法层面没问题时Python 代码里的问题通常就集中在类型转换、连接参数和筛选条件上。先把单连接、单条增删改查跑稳再碰索引、聚合和批量写入这条路我建议你也走一遍。