这一篇是“100天精通Python”系列的第 40 天主题非常直接用pymongo操作 MongoDB 数据库从安装、连接到增删改查再到批量任务和索引优化最后带上接口封装思路和常见问题排查。MongoDB 是目前最常用的文档型 NoSQL 数据库数据以 BSON 格式存储业务上可以理解成“存在数据库里的 JSON 对象”。Python 操作 MongoDB 最主流的方案就是官方驱动pymongo它的 API 设计比较简洁和 MongoDB 的 Shell 语法很接近适合做爬虫存储、日志收集、用户行为分析、配置中心这类场景。本文不绕弯子直接分成几个部分先看 pymongo 的核心能力、再准备环境、然后启动 MongoDB 服务、接着用代码把增删改查和批量任务全部跑一遍最后给出索引优化、接口 API 封装和常见坑的排查方法。如果你正在学 Python 操作数据库或者准备把项目里的数据存储切到 MongoDB这篇文章可以直接作为操作手册。1. 核心能力速览能力项说明项目类型Python 官方 MongoDB 驱动程序核心功能连接 MongoDB、数据库/集合管理、增删改查、聚合管道、索引管理、批量写入、事务支持支持平台Windows、Linux、macOSPython 版本建议 Python 3.8 及以上具体以 pymongo 当前版本要求为准MongoDB 版本支持 MongoDB 3.6 及以上新版 pymongo 建议配合 MongoDB 4.2 使用安装方式pip install pymongo通信方式TCP默认端口 27017批量能力insert_many、bulk_write、游标批量遍历接口 API配合 Flask/FastAPI 可封装成 HTTP 接口适合场景爬虫数据存储、日志收集、用户行为记录、配置中心、快速原型开发不适合场景复杂多表关联查询、强事务一致性、图关系分析从材料看pymongo 的定位是“用 Python 的方式操作 MongoDB”所以你在 MongoDB Shell 里能做的操作在 pymongo 里基本都能对应着写出来。安装成本很低没有额外的编译依赖只要有网络环境一条命令就能装好。2. 适用场景与使用边界先判断这个工具适不适合你。MongoDB 的文档模型比关系型表结构更灵活一个集合collection里的文档不要求字段完全一致新增字段不需要先改表结构。这种特性非常适合数据格式多变、字段可能长期演进的业务比如爬虫保存的页面数据、埋点日志、用户配置信息。pymongo 在这种场景下能把 Python 字典直接写入数据库读出来也是字典和 Python 对象之间的转换成本很低。但也要注意边界。如果你需要大量JOIN操作、多行事务、复杂聚合统计MongoDB 虽然支持一部分但并不是最优选择。传统 ERP、金融账务系统还是建议用 MySQL 或 PostgreSQL。pymongo 只是数据库驱动它本身不会帮你解决数据建模问题集合设计不好同样的查询会明显变慢。使用边界方面重点提三点不要把 MongoDB 服务直接暴露到公网必须开启访问认证否则很容易被扫描和写入恶意数据。涉及用户隐私、电话号码、身份信息等敏感字段建议在写入前加密或脱敏至少不要明文保存在没有权限控制的集合里。要定期备份数据特别是使用副本集或单机部署时否则误操作delete_many后恢复成本很高。3. 环境准备与前置条件在写代码之前需要把 Python、MongoDB、pymongo 三样东西准备好。3.1 安装 MongoDBMongoDB 需要单独安装。Windows 用户可以下载 MongoDB Community Server 的 MSI 安装包安装时选择“Install MongoDB as a Service”可以注册成 Windows 服务开机自启。Linux 用户可以通过 apt 或 yum 安装macOS 用户可以用 Homebrew。安装完成后确认mongod命令可用。Windows 下如果配置了服务可以直接启动服务如果没配置可以手动指定数据目录启动# Linux/macOS 示例需要先创建数据目录 mkdir -p /data/db mongod --dbpath /data/db --port 27017Windows 命令行启动方式mongod --dbpath D:\mongodb\data --port 27017启动日志中出现Waiting for connections说明服务已经正常监听 27017 端口。如果只想快速验证也可以不用本地安装直接在测试服务器上用 Docker 启动docker run -d --name mongodb -p 27017:27017 mongo:6.0mongo:6.0是示例镜像标签具体版本可以根据自己的环境调整。3.2 安装 Python 和 pymongoPython 建议用 3.8 以上版本。先确认 Python 环境python --version然后安装 pymongopip install pymongo如果需要操作 MongoDB 的聚合管道、GridFS、加密等扩展功能还可以安装pymongo[srv]或pymongo[encryption]但基础使用只需要一个pymongo包。验证安装是否成功python -c import pymongo; print(pymongo.version)能输出版本号说明驱动已经可用。3.3 检查 MongoDB 服务状态在 Python 中连接之前建议先用命令行工具mongosh或者直接通过 pymongo 检查。最简单的检查方式是写一段连接代码后面会讲到。只要端口没被防火墙拦截连接本身不会太复杂。4. 连接 MongoDB 数据库连接 MongoDB 的核心类是MongoClient。这一步要解决两件事如何连接本地服务、如何带认证信息连接。4.1 创建 MongoClient 连接先把代码写出来from pymongo import MongoClient # 本地默认端口连接 client MongoClient(mongodb://127.0.0.1:27017/) # 获取数据库不存在会自动创建 db client.mydb # 获取集合不存在会自动创建 collection db.users这里client.mydb和client[mydb]是等价的。MongoDB 是“懒创建”机制只有实际执行写入操作时数据库和集合才会真正创建。4.2 带用户名密码的连接如果 MongoDB 开启了认证就需要在连接串里带上账号client MongoClient( mongodb://admin:password123127.0.0.1:27017/?authSourceadmin )authSource表示认证库常见为admin。如果你的账号建立在某个业务库下authSource就写对应的库名。生产环境建议用环境变量保存连接串不要硬编码到代码里。4.3 连接参数MongoClient还支持一些常用参数client MongoClient( mongodb://127.0.0.1:27017/, serverSelectionTimeoutMS5000, # 5秒内选不到服务器就报错 connectTimeoutMS3000, maxPoolSize50 )maxPoolSize控制连接池大小。默认的MongoClient是懒连接真正发起操作时才会建立连接。如果你在脚本里只是创建 client 并不操作它不会立刻报错。4.4 验证连接是否成功from pymongo import MongoClient client MongoClient(mongodb://127.0.0.1:27017/, serverSelectionTimeoutMS5000) # 向服务器发送 ping 命令 try: client.admin.command(ping) print(MongoDB 连接成功) except Exception as e: print(连接失败:, e)这个ping命令是验证网络和服务状态最直接的方式。如果失败先检查服务有没有启动、端口是否正确。5. 文档插入操作MongoDB 的基本数据单元是文档在 pymongo 中对应 Python 字典。插入操作有insert_one和insert_many。5.1 插入单条文档from pymongo import MongoClient from datetime import datetime client MongoClient(mongodb://127.0.0.1:27017/) db client.mydb collection db.users user { name: 张三, age: 28, email: zhangsanexample.com, tags: [python, mongodb], created_at: datetime.now() } result collection.insert_one(user) print(result.inserted_id)插入成功后result.inserted_id会返回自动生成的ObjectId。如果你在文档里手动指定了_id字段MongoDB 会使用你指定的值。5.2 批量插入文档users [ {name: 李四, age: 32, city: 上海}, {name: 王五, age: 24, city: 北京}, {name: 赵六, age: 29, city: 广州} ] result collection.insert_many(users) print(result.inserted_ids)insert_many接受一个列表返回一个包含所有_id的列表。批量插入比循环insert_one效率高很多因为减少了客户端和服务端的往返次数。5.3 插入数据后的验证count collection.count_documents({}) print(当前用户数量:, count)注意count()方法在新版 pymongo 中已经弃用统一使用count_documents。6. 文档查询操作查询是 pymongo 里用得最多的部分。掌握查询语法基本就能解决 80% 的日常需求。6.1 查询单条文档find_oneuser collection.find_one({name: 张三}) print(user)返回的是字典找不到时返回None。和 MongoDB Shell 的findOne行为一致。6.2 查询多条文档findcursor collection.find({age: {$gte: 25}}) for user in cursor: print(user[name], user[age])find返回一个Cursor对象是一个可迭代的游标。可以用list()转换成列表但如果数据量很大不建议一次性全部加载到内存。6.3 常用比较条件操作符含义示例$eq等于{age: {$eq: 28}}$ne不等于{age: {$ne: 28}}$gt大于{age: {$gt: 25}}$gte大于等于{age: {$gte: 25}}$lt小于{age: {$lt: 30}}$lte小于等于{age: {$lte: 30}}$in在列表中{city: {$in: [北京, 上海]}}$nin不在列表中{city: {$nin: [北京]}}$regex正则匹配{name: {$regex: ^张}}# 查询年龄在 25 到 30 之间或者城市为上海的用户 query { $or: [ {age: {$gte: 25, $lte: 30}}, {city: 上海} ] } for user in collection.find(query): print(user)6.4 排序、分页、计数# 按年龄降序排序 cursor collection.find().sort(age, -1) # 跳过前2条只取3条 cursor collection.find().sort(age, -1).skip(2).limit(3) # 只返回 name 和 age 字段不返回 _id cursor collection.find({}, {_id: 0, name: 1, age: 1})skip加limit是传统分页方案。数据量超过几万页时更推荐用_id或排序字段做游标分页避免大偏移量导致性能下降。6.5 正则和数组查询# 名字以 张 开头 users collection.find({name: {$regex: ^张}}) # tags 数组中包含 python users collection.find({tags: python}) # tags 数组同时包含 python 和 mongodb users collection.find({tags: {$all: [python, mongodb]}})数组查询在标签系统、权限系统里很常用$all表示数组字段必须同时包含多个值。7. 文档更新操作更新操作最常用的是update_one和update_many。更新时要注意直接传入整个新文档会替换原文档更新指定字段必须使用更新操作符比如$set、$inc、$unset、$push、$addToSet。7.1 修改指定字段update_oneresult collection.update_one( {name: 张三}, {$set: {age: 29, city: 深圳}} ) print(result.matched_count, result.modified_count)matched_count表示匹配到的文档数modified_count表示实际被修改的文档数。这里能发现如果新值和旧值一样modified_count是 0。7.2 批量更新update_many# 将所有北京用户的年龄加 1 result collection.update_many( {city: 北京}, {$inc: {age: 1}} ) print(result.modified_count)$inc是数值增减操作适合计数、积分、库存等场景。7.3 数组更新$push和$addToSet# 向 tags 数组追加一个值 collection.update_one( {name: 张三}, {$push: {tags: pymongo}} ) # 如果值已经存在则不重复添加 collection.update_one( {name: 张三}, {$addToSet: {tags: pymongo}} )$push每次都会追加$addToSet会去重。需要维护标签、关注列表、操作日志时优先考虑$addToSet。7.4 替换整条文档replace_onenew_doc { name: 张三, age: 30, city: 广州, tags: [python] } collection.replace_one({name: 张三}, new_doc)replace_one会删除原文档中其他字段用新文档整体替换。除非业务明确需要否则更推荐$set做局部更新避免误删字段。7.5 upsert 更新或插入result collection.update_one( {name: 孙七}, {$set: {age: 18, city: 杭州}}, upsertTrue ) print(result.upserted_id)加上upsertTrue后如果查询条件匹配不到文档就会插入一条新文档。这个能力在“存在则更新不存在则创建”的业务中非常实用。8. 文档删除操作删除操作分为delete_one和delete_many。# 删除名字为张三的第一条记录 result collection.delete_one({name: 张三}) print(result.deleted_count) # 删除所有城市为北京的用户 result collection.delete_many({city: 北京}) print(result.deleted_count) # 清空集合保留集合本身 collection.delete_many({})删除整个集合可以用collection.drop()清除整个数据库client.drop_database(mydb)删除是不可逆操作建议在执行delete_many前先用count_documents看一下确定影响范围或者在测试库上先验证。9. 索引与批量任务索引是 MongoDB 查询性能的核心。没有索引时MongoDB 必须做全表扫描数据到几十万条后查询延迟会明显上升。9.1 创建索引# 单字段索引 collection.create_index(name) # 复合索引 collection.create_index([(city, 1), (age, -1)]) # 唯一索引 collection.create_index(email, uniqueTrue) # 查看集合索引 for index in collection.list_indexes(): print(index)1表示升序-1表示降序。如果业务查询经常同时按city和age过滤排序适合建复合索引。唯一索引可以用来防止邮箱、手机号等字段重复。9.2 批量写入与bulk_write批量任务指的是在一次请求里执行多条写操作。前面提到的insert_many只能批量插入而bulk_write可以混合插入、更新、删除。from pymongo import InsertOne, UpdateOne, DeleteOne operations [ InsertOne({name: 测试1, age: 20}), UpdateOne({name: 测试2}, {$set: {age: 21}}, upsertTrue), DeleteOne({name: 测试3}) ] result collection.bulk_write(operations) print(result.inserted_count) print(result.modified_count) print(result.deleted_count)bulk_write适合做数据同步、清洗、ETL 任务。它默认按顺序执行如果中间某条失败后面的操作可能不会继续可以使用orderedFalse让 MongoDB 继续执行剩余操作result collection.bulk_write(operations, orderedFalse)9.3 批量查询与游标遍历如果要对全量数据做处理不要一次性list()出来应该使用游标分块遍历batch_size 500 cursor collection.find({}).batch_size(batch_size) for doc in cursor: # 处理每条文档 process(doc)batch_size控制每次从服务端取回的文档数量可以避免把大量数据一次性加载到内存。10. 接口 API 调用示例pymongo 本身不是 HTTP 服务但实际项目中经常需要把数据库操作封装成接口让前端或内部服务调用。下面用 Flask 做一个最简单的示例演示如何把 pymongo 继承到 API 层。pip install flask pymongofrom flask import Flask, request, jsonify from pymongo import MongoClient from bson import ObjectId app Flask(__name__) client MongoClient(mongodb://127.0.0.1:27017/) db client.mydb collection db.users def serialize_doc(doc): 把 ObjectId 转成字符串方便 JSON 返回 if doc is None: return None doc[_id] str(doc[_id]) return doc app.route(/users, methods[GET]) def list_users(): users list(collection.find({})) return jsonify([serialize_doc(user) for user in users]) app.route(/users, methods[POST]) def create_user(): data request.get_json() if not data or not data.get(name): return jsonify({error: name is required}), 400 result collection.insert_one(data) return jsonify({_id: str(result.inserted_id)}), 201 app.route(/users/user_id, methods[PUT]) def update_user(user_id): data request.get_json() result collection.update_one( {_id: ObjectId(user_id)}, {$set: data} ) if result.matched_count 0: return jsonify({error: user not found}), 404 return jsonify({message: updated}) app.route(/users/user_id, methods[DELETE]) def delete_user(user_id): result collection.delete_one({_id: ObjectId(user_id)}) if result.deleted_count 0: return jsonify({error: user not found}), 404 return jsonify({message: deleted}) if __name__ __main__: app.run(host127.0.0.1, port5000, debugTrue)这是一个通用示例实际部署时要注意接口必须做权限校验不能把数据库接口直接裸露给公网。入参要校验避免用户传入$set之类的内容修改非预期字段。ObjectId转换失败时要做异常处理。大规模接口服务建议使用 FastAPI Pydantic做自动化参数校验和文档输出。也可以用curl快速验证接口curl http://127.0.0.1:5000/userscurl -X POST http://127.0.0.1:5000/users \ -H Content-Type: application/json \ -d {name: 接口用户, age: 18}11. 资源占用与性能观察MongoDB 是内存友好的数据库吗实际上它依赖操作系统缓存索引和热数据尽可能放在内存中查询速度很快。单机部署时MongoDB 进程会占用较多内存这是正常现象因为 LRU 缓存会主动利用空闲内存。观察资源占用时重点看以下指标观察项方法说明MongoDB 内存占用任务管理器或mongostat缓存大小会动态变化连接数db.serverStatus().connections连接池如果耗尽请求会阻塞查询耗时explain(executionStats)检查是否走索引磁盘 IOiostat或云监控写入量大时磁盘容易成为瓶颈Python 端连接池maxPoolSize高并发场景要合理设置explain是排查慢查询的重要工具cursor collection.find({city: 北京}).sort(age, -1) explain_result cursor.explain() print(explain_result[queryPlanner][winningPlan])如果 winning plan 显示COLLSCAN说明没有走索引。这时应该考虑建立合适的索引。使用 pymongo 时有几个性能习惯一个进程内复用同一个MongoClient不要每次操作都新建连接。批量插入优先insert_many不要循环insert_one。查询时只返回需要的字段减少网络传输。避免一次find后把所有结果转成 list改用游标。写入量大的业务开启确认写w1即可不要用wmajority做不必要的确认除非你确实需要强一致。12. 常见问题与排查方法问题现象可能原因排查方式解决方案连接超时ServerSelectionTimeoutErrorMongoDB 服务未启动、端口错误、防火墙拦截检查mongod进程、netstat -ano | findstr 27017、Telnet 端口启动服务确认连接串端口放行防火墙认证失败Authentication failed用户名密码错误、认证库不对、用户没有权限用mongosh测试认证信息确认authSource指向正确的认证库重新设置用户插入文档报DocumentTooLarge单条文档超过 16 MB 限制查看报错内容压缩字段、拆分文档或使用 GridFS 存储大文件count()报错或告警新版 pymongo 移除了旧count查看版本变更改用count_documents({})查询速度很慢没有索引或索引不合理执行explain查看COLLSCAN创建单字段或复合索引批量写入部分失败orderedTrue模式下中途出错检查报错开启日志使用orderedFalse或对错误记录重试连接数过多每次操作创建新MongoClient且未复用查看连接数复用全局 client调整maxPoolSize数据乱码字符集问题检查写入编码客户端统一使用 UTF-8删除后数据无法恢复没有备份检查备份策略使用mongodump定期备份部署副本集最常见的问题是“明明本地装了 MongoDB但 Python 连不上”绝大多数原因是 MongoDB 服务没有启动或者连接串端口写错。建议先跑一下ping命令再排查其他环节。13. 最佳实践与使用建议pymongo 的 API 虽然简单但项目落地时还是有一些工程化经验可以提前规避问题。第一连接管理要统一。建议在项目入口创建全局MongoClient通过配置模块读取连接串不要在每个函数里都MongoClient(...)。这样既能复用连接池也方便切测试库和生产库。# config.py import os from pymongo import MongoClient MONGO_URI os.getenv(MONGO_URI, mongodb://127.0.0.1:27017/) client MongoClient(MONGO_URI) db client.mydb第二集合名和字段名要有规范。集合名建议使用复数名词比如users、orders。字段名统一使用小写和下划线避免出现userId和user_id混用。字段名风格不统一后期写查询条件时会非常痛苦。第三建立索引要有计划。不要上来就给所有字段加索引索引会占用磁盘空间并且降低写入性能。应该先分析查询条件对高频过滤字段建立索引对排序字段建立复合索引。第四生产环境一定要开启认证和访问控制。MongoDB 默认是没有账号密码的如果监听在0.0.0.0相当于把数据裸奔在公网上。至少要做两件事服务监听内网 IP创建专用业务账号并分配最小权限。第五涉及敏感数据时要加密和脱敏。不能把email、phone、id_card这类字段明文存储除非数据库网络和权限完全隔离。建议写入前用加密算法处理展示时再解密或者完成业务后自动清理。第六批量任务要加上日志和失败重试机制。用bulk_write处理大量数据时如果中间出错最好捕获异常并记录失败的文档 key后续单独重试。不要在一次任务里写入上万条还不打印任何日志出了问题很难定位。第七备份策略要提前设计。至少使用mongodump做定时全量备份有条件的部署副本集实现高可用。对于核心业务还可以开启 op log用增量备份降低数据丢失风险。14. 总结与下一步这一篇把 Python 操作 MongoDB 的完整路径过了一遍环境准备、服务启动、pymongo 连接、插入、查询、更新、删除、索引、批量任务、接口封装和排错方法。最值得先动手验证的是insert_onefind_oneupdate_onedelete_one这条基础链路跑通之后MongoDB 的核心用法你就已经掌握了。最容易踩的坑有三个第一MongoDB 服务没启动就急着跑 Python 代码结果报连接超时第二更新文档时忘记用$set结果整条文档被覆盖第三大量数据查询时没有建索引数据量上去之后接口变慢。如果这周有时间建议继续往下练几个方向MongoDB 聚合管道aggregate做分组统计GridFS存大文件副本集事务操作以及motor异步驱动对接 FastAPI。等基础增删改查熟练之后再去看聚合和事务会轻松很多。建议把这篇文章收藏备用下次需要操作 MongoDB 时可以直接对照着写代码。