企业官网建设流程全解析

上周有个朋友问我，为什么自己用 Python 连 MongoDB 的脚本总是卡在第一步，不是连接超时，就是查出来的东西跟预期不一样。我说你大概率是把 MongoDB 当 MySQL 用了，尤其是那几个最基础的增删改查，看起来简单，实际坑不少。这篇文章是 Python 使用 MongoDB 系列的第一篇，我先把环境连接、核心增删改查、索引优化和常见排错这四块讲透，内容偏实战，适合刚接触 PyMongo 的读者，也适合已经在写业务代码但偶尔被性能或异常卡住的同学。

我这些年用 Python 操作 MongoDB，主要是在做数据采集和用户行为分析类的内部系统。这类业务的文档结构不固定、字段经常变，用关系库会改表改到怀疑人生，MongoDB 的文档模型反而很省心。但省心不代表没门槛，很多问题都出现在“会用了”和“用对了”之间的那条缝里。比如同一个查询条件，索引建没建，响应时间可能差几十倍；同一个 update，写没写 $set，文档结构可能直接被改坏。这篇就把这些基础但关键的细节一个个拆开。

1. 环境准备与连接参数

1.1 驱动选型与版本选择

Python 操作 MongoDB 最主流的驱动是 PyMongo，MongoDB 官方维护，功能覆盖完整，文档也全。如果你的 Web 框架是 FastAPI、Tornado 这类异步框架，可以选 Motor，它是 PyMongo 的异步封装。但普通脚本、批处理任务、同步 Web 服务，直接用 PyMongo 就够了，别一上来就整异步，异步带来的心智负担和排查成本远大于它节省的那点时间。

安装很简单：

pip install pymongo

不过我建议在项目里固定版本，比如pip install pymongo==4.6.3。PyMongo 各版本对 MongoDB 服务端的兼容性有差异，新版本可能会引入一些行为变化。我线上项目一般不会盲目追求最新版，而是锁定一个经过验证的版本，然后用 requirements.txt 或 pyproject.toml 管起来。升级驱动前至少要在测试环境跑一遍全量回归，尤其是涉及聚合管道和地理查询的项目。

1.2 连接字符串与会话参数

连接本地 MongoDB 实例，最直接的方式是这样：

from pymongo import MongoClient from pymongo import serverSelectionTimeoutMS client = MongoClient( "mongodb://127.0.0.1:27017", username="admin", password="your_password", authSource="admin", serverSelectionTimeoutMS=3000, )

这里有两个容易被新手忽略的点。第一个是authSource。MongoDB 在认证时，默认去admin库查找用户，如果你的用户是在admin库创建的，那authSource="admin"是对的；如果用户是在某个业务库里创建的，就必须写成对应的库名。很多“账号密码明明正确却报 Authentication failed”的案例，最后查下来都是authSource写错。

第二个是serverSelectionTimeoutMS。MongoClient 是惰性连接，真正建立连接是在第一次数据库操作时。如果不显式设置这个超时时间，默认情况下客户端会花很长时间去探测服务器，可能等 30 秒才抛错，这在脚本里非常折磨人。我习惯设置成 3000 毫秒甚至更短，让问题快速暴露，而不是等到超时。

1.3 连接池与客户端复用

很多人会犯一个错误：每次操作数据库都 new 一个 MongoClient，用完又丢弃。这是非常糟糕的用法。MongoClient 内部维护了一个连接池，正确的姿势是让它在进程内保持单例，所有模块共用同一个客户端实例。

我在项目里通常会在一个单独的模块里创建客户端，然后其他模块直接 import 这个实例：

# db.py from pymongo import MongoClient client = MongoClient("mongodb://127.0.0.1:27017", serverSelectionTimeoutMS=3000) db = client["app_db"]

这样不仅快，还能避免把 MongoDB 的连接数打满。如果业务脚本是短任务，比如跑一个 cron 批处理，那每次启动新进程、新建客户端是合理的，但进程退出前最好显式调用client.close()释放连接。

2. 核心增删改查操作的真正写法

2.1 插入文档与 _id 的生成规则

拿到集合对象之后，最基本的插入操作是这样：

users = db["users"] doc = { "name": "张三", "age": 28, "tags": ["python", "mongo"], "created_at": datetime.now(timezone.utc), } result = users.insert_one(doc) print(result.inserted_id)

MongoDB 在插入时如果文档里没有_id字段，会自动生成一个ObjectId对象。ObjectId 不是一个简单的随机字符串，它由 12 个字节组成：前 4 字节是时间戳，中间 5 字节是机器标识，接着 2 字节是进程 ID，最后 3 字节是自增计数器。这也是为什么看起来_id好像按插入顺序排列，但实际上在多进程、多机器环境下并不保证单调递增。如果业务上需要严格的插入顺序，建议自己额外维护一个自增字段或时间字段，不要依赖_id排序。

批量插入用insert_many：

docs = [ {"name": f"user_{i}", "age": i % 50, "created_at": datetime.now(timezone.utc)} for i in range(1000) ] result = users.insert_many(docs, ordered=False) print(len(result.inserted_ids))

ordered参数值得单独说。默认ordered=True，意思是按顺序逐条插入，如果中途某条失败，后面的全部不执行。如果ordered=False，MongoDB 会尝试继续插入后面的文档，适合数据批处理场景。但要注意，ordered=False时错误不会立刻中断，脚本会继续跑，所以要用insert_many_result检查实际插入了多少条，避免数据静默丢失。

2.2 查询文档与游标的使用

查询单个文档用find_one，查询多个文档用find。find返回的是一个游标对象，不是数组。这个区别很关键。

user = users.find_one({"name": "张三"}) print(user) cursor = users.find({"age": {"$gte": 18, "$lt": 30}}) for u in cursor: print(u)

游标是惰性的，它不会一次性把所有数据加载到内存，而是按需向服务器索取下一批数据。这在大数据量查询时是好事，但也会带来几个坑。

第一个坑是：不要在遍历游标的同时修改集合。比如你在一个循环里对查询出来的文档做 update，可能会导致结果不可预期。我踩过一次，当时是在一个日志导入脚本里，遍历游标时对每条记录做去重更新，结果发现部分文档被跳过了。这不是 MongoDB 在偷偷丢数据，而是游标在迭代过程中，集合数据变化影响了后续 batch 的结果。解决方法是先把_id列表取出来，再分批处理。

第二个坑是：如果只用了游标的一部分，要记得释放游标。操作简单，在不需要的时候调用cursor.close()。不过 Python 的for循环遍历完会自动释放，只有提前break的情况下才需要手动处理。

查询条件的写法也有讲究。比较运算直接用$gte、$lte、$ne这些操作符，正则查询用$regex：

users.find({"name": {"$regex": "^张"}})

数组字段可以用$in、$all、$elemMatch来筛选：

# tags 中包含 "python" 或 "mongo" users.find({"tags": {"$in": ["python", "mongo"]}}) # tags 中同时包含 "python" 和 "mongo" users.find({"tags": {"$all": ["python", "mongo"]}}) # 数组元素为对象时的复杂匹配 users.find({"tags": {"$elemMatch": {"name": "python", "level": "advanced"}}})

控制返回字段用投影参数，第二个参数传入一个字典，1 表示返回，0 表示不返回：

users.find({"age": {"$gte": 18}}, {"_id": 0, "name": 1, "age": 1})

注意_id默认是返回的，如果不想返回它，要显式指定"_id": 0。

2.3 更新文档与字段操作符

更新文档最容易翻车，很多人直接用 MySQL 的思维来写：

users.update_one({"_id": ObjectId("...")}, {"age": 31})

这样写不会报错，但结果很惨：它把整个文档替换成了只包含age字段的文档，其他字段全部丢失。MongoDB 的 update 默认是“文档替换”语义，不是“字段修改”语义。要修改字段，必须用字段更新操作符。

我的标准写法是这样的：

users.update_one( {"name": "张三"}, { "$set": {"age": 29}, "$inc": {"login_count": 1}, "$push": {"history": {"time": datetime.now(timezone.utc), "action": "login"}}, }, )

$set用于设置或覆盖字段，$inc用于数值自增，$push用于向数组追加元素。还有$pull用于从数组移除元素，$unset用于删除字段。这些操作符可以组合使用，一次性完成多个字段的变更。

update_many用于批量更新，比如给所有未成年用户打标记：

users.update_many( {"age": {"$lt": 18}}, {"$set": {"status": "minor"}} )

还有一个很容易被忽略的参数是upsert。当匹配条件找不到文档时，upsert=True会插入一条新文档，新文档由匹配条件和$set里的字段组合而成。这个特性在做计数器、去重表、日报聚合时非常好用，比如按天记数：

daily_stats.update_one( {"date": today_str}, {"$inc": {"pv": 1, "uv": 1}}, upsert=True, )

2.4 删除文档与数据安全

删除操作也一样，delete_one删除第一条匹配的文档，delete_many删除所有匹配的文档。很多人会把这两个搞混，尤其是在脚本里想删一批测试数据，却调用了delete_many，或者反过来。

users.delete_one({"name": "张三"}) users.delete_many({"status": "inactive"})

如果想把整个集合和它的索引都删掉，用drop()：

users.drop()

我强烈建议在删除操作前先执行一次查询，确认匹配数量：

count = users.count_documents({"status": "inactive"}) print(f"准备删除 {count} 条，确认继续？")

在业务系统里，删除数据前最好先想到能不能用逻辑删除。我常用的做法是给集合加一个deleted字段，查询时自动过滤deleted=False，更新时只修改状态。这样即使误操作了，数据还在物理留存，可以从容地排查和恢复。尤其对用户行为日志、订单这类数据，物理删除的成本远比保留状态高。

3. 索引与查询性能优化

3.1 单字段索引与执行计划

MongoDB 默认只在_id上建索引。如果你的查询条件里用的是其他字段，比如{"status": "active"}，那 MongoDB 就得做全集合扫描，数据量小的时候无所谓，数据量到几百万条后，一次全表扫描能让你怀疑服务是不是挂了。

创建索引很简单：

users.create_index([("status", 1)])

索引方向 1 或 -1 在这个字段上是等值的场景下没有区别，但对排序场景有影响。比如经常用created_at倒序查最新记录，就应该建一个降序索引：

users.create_index([("created_at", -1)])

怎么确认查询有没有走索引？用 explain 看执行计划：

explain_result = db.command( "explain", users.find({"status": "active"}), verbosity="executionStats", ) print(explain_result["queryPlanner"]["winningPlan"])

如果你看到COLLSCAN，说明是集合扫描，没走索引；看到IXSCAN，说明走的是索引扫描。这个判断在优化阶段非常有用。我一般会写一个小函数把执行计划里的关键信息打出来，包括totalDocsExamined和totalKeysExamined，两个数差得越远，说明索引利用率越低。

3.2 复合索引与最左前缀规则

实际业务里，单字段索引往往不够用。比如常见查询是“状态为 active 的用户，按创建时间倒序，分页拉取”，这个条件涉及status和created_at两个字段，单字段索引只能帮你过滤掉一部分数据，性能远不如复合索引。

创建复合索引：

users.create_index([("status", 1), ("created_at", -1)])

复合索引生效遵循最左前缀规则。查询条件里必须包含最左边的字段，索引才会被使用。比如上面这个复合索引，可以支撑{"status": "active"}这个单独条件的查询，也可以支撑{"status": "active", "created_at": {"$gte": ...}}的查询，但如果查询条件里只有created_at，这个索引就不起作用。

我设计复合索引的经验顺序是：等值匹配字段放最前面，排序字段放中间，范围查询字段放后面。比如查询里同时有“按用户 id 等值过滤”和“按时间范围过滤”，索引就应该建在[("user_id", 1), ("created_at", -1)]上。这是实践中最常用也最不容易错的顺序。

3.3 索引的代价与维护

索引不是越多越好。每建一个索引，写入时就要额外维护一颗 B 树，写放大是实打实的。我见过一个团队为了追求查询速度，把每个可能的筛选字段都建了索引，结果插入性能掉了接近一半，磁盘占用也涨了不少。

建议把线上用不到的索引定期清理掉。查看集合现有索引：

print(users.index_information())

删除某个索引：

users.drop_index("status_1")

删除全部非_id索引：

users.drop_indexes()

还有一个小技巧：大集合建索引尽量在低峰期执行，并且用后台方式创建。虽然新版 MongoDB 默认是可以在线建索引的，但超大集合建索引期间依然会有资源和锁竞争，最好评估好窗口再动手。

4. 常见问题排查与避坑

4.1 连接超时与认证失败

遇到ServerSelectionTimeoutError: No server candidates时，按这个顺序排查：

• 服务有没有启动？用mongosh --eval "db.runCommand({ping:1})"先测一下。
• IP 和端口对不对？是否被防火墙挡了？
• 认证库对不对？重点检查authSource。
• 是否能被其他客户端访问？同一台机器上有没有网络隔离？

我在运维一个内部工具时遇到过一种诡异情况：脚本在本地跑得好好的，部署到容器里就连不上，排查下来是容器网络的 DNS 解析问题，MongoDB 域名解析到了旧 IP。这种问题看错误信息里的 candidate 地址会帮你快速定位。

4.2 ObjectId 与时间处理

ObjectId 可以直接转字符串用，也可以从里面提取时间：

from bson import ObjectId oid = ObjectId("657c1f2a5a5a1a1a1a1a1a") print(oid.generation_time)

这里有个需要注意的细节：generation_time是 naive 的 UTC 时间，不带时区信息，要和本地时间对比时自己手动加上时区。我建议所有写入 MongoDB 的时间字段都统一用 UTC 时间存储，展示层再转本地时区。否则你会遇到一个很经典的问题：白天写入的数据到了晚上查起来“消失”了几个小时，或者报表里数据多出来 8 小时偏移。真实原因就是时区不一致。

PyMongo 写 datetime 字段时，推荐用：

from datetime import datetime, timezone datetime.now(timezone.utc)

不要用datetime.utcnow()，它在 Python 3.12 之后已经被标记为 deprecated，而且它返回的是 naive UTC 时间，在很多框架里容易和本地时间混用。

4.3 文档大小限制与批量操作限制

MongoDB 单条 BSON 文档最大是 16MB。这比 MySQL 的单行限制宽松得多，但也不是无限大。如果业务里有存大文件、大日志的场景，建议把超过 1MB 的内容拆分到 GridFS 或对象存储，不要在集合里硬塞。

insert_many一次允许的文档数量上限是 100000，而且写入数据包总大小不能超过约 48MB 左右（不同版本略有差异）。大批量导入时我习惯分片写入：

batch_size = 5000 for i in range(0, len(data), batch_size): chunk = data[i : i + batch_size] collection.insert_many(chunk, ordered=False) print(f"已写入 {i + len(chunk)} / {len(data)}")

4.4 写关注与数据安全

MongoDB 的写操作默认是 acknowledged 模式，服务端确认写入成功后才返回。但客户端可以配置更弱的写关注，比如w=0，表示不等待确认。w=0确实能让写入更快，但代价是如果服务器在那一刻出了问题，你根本不知道数据写没写进去。日志类、非关键数据可以权衡，但用户数据、支付数据绝对别用w=0。

在副本集环境中，我常常用w="majority"，表示需要大多数副本节点确认写入，这样即使某个节点故障，数据也不容易丢。代价是每次写操作要等待副本同步，延迟会高一点。生产环境要根据业务对数据安全的需求来权衡，不是所有业务都适合 majority。

4.5 循环查询与游标快照问题

前面提到过遍历游标时修改集合是有风险的。这里再补充一个常见场景：在for doc in collection.find(...)的循环里对同一个集合做update_one。我第一次遇到时排查了很久，后来发现游标在迭代过程中，如果集合数据发生变化，MongoDB 可能重新拉取新数据，导致同一批文档被处理多次或不一致。

保险做法是先把_id提取到一个列表里：

ids = [doc["_id"] for doc in collection.find({"status": "pending"}, {"_id": 1})] for oid in ids: collection.update_one({"_id": oid}, {"$set": {"processed": True}})

这样游标遍历完整结束之后再做写入，数据和状态都是可控的。

4.6 问题速查表

这个系列第一篇就写到这里。我自己的体会是，PyMongo 的 API 确实简单，真正的坑往往在驱动之外——连接参数的语义、ObjectId 的时间属性、复合索引的字段顺序、写关注的选择，每一个不注意都可能变成线上的麻烦。下一篇文章我准备拆解聚合管道aggregate和$lookup的实战写法，那部分才是 MongoDB 真正体现优势的地方。如果你在实操中也遇到过一些看起来玄学的问题，欢迎带着错误信息来讨论，我们一起把坑填平。