优先使用pymongo而非flask-pymongo,因其暴露连接池、超时、异常等关键配置;需显式设置serverselectiontimeoutms等参数防卡死;objectid须在业务层序列化;集合需手动建索引与校验。

PyMongo 是直接、可控的 MongoDB Python 驱动,Flask-PyMongo 是封装层——优先用 PyMongo,除非你明确需要 Flask 上下文自动管理连接生命周期。
为什么不用 Flask-PyMongo?
它把 PyMongo 包了一层,但隐藏了连接池配置、超时控制和异常细节,导致问题难定位。比如:
-
mongo.db.users.find()看似简洁,但底层连接复用逻辑不透明,高并发下容易耗尽连接 - 配置项如
MONGO_MAX_POOL_SIZE在 Flask-PyMongo 中默认是 100,但没文档说明它是否生效 - 当 MongoDB 服务临时不可达时,
PyMongo会抛ConnectionFailure,而 Flask-PyMongo 可能静默重试或卡住
真实项目里,你更常需要:手动控制连接、设置 serverSelectionTimeoutMS、捕获 ServerSelectionTimeoutError 做降级,这些 PyMongo 原生支持,Flask-PyMongo 不暴露。
PyMongo 连接必须显式配置超时和重试
默认连接行为在生产环境极危险:无超时、无限重试、不校验服务器状态。
正确初始化方式:
from pymongo import MongoClient from pymongo.errors import ServerSelectionTimeoutError <p>client = MongoClient( "mongodb://localhost:27017/", serverSelectionTimeoutMS=3000, # 连接发现超时 3 秒 connectTimeoutMS=5000, # TCP 连接超时 5 秒 socketTimeoutMS=10000, # 每次读写操作超时 10 秒 maxPoolSize=50, # 连接池上限,别用默认 100(内存/端口压力) minPoolSize=5, # 预热连接数,避免冷启动延迟 retryWrites=True, # 启用写重试(4.0+ 版本必需) )</p>
漏掉 serverSelectionTimeoutMS,一旦 MongoDB 宕机,find() 会卡住整整 20 秒(PyMongo 默认值);漏掉 retryWrites,副本集切换时可能丢写入。
ObjectId 转换必须在业务层做,不能依赖视图模板
ObjectId 无法被 JSON 直序列化,常见错误:TypeError: Object of type ObjectId is not JSON serializable。
Python 3.14.2是Python编程语言在2025年12月5日发布的稳定版本,属于3.14系列的第二个维护更新。该版本包含了18项修复,重点解决了多进程、数据类及正则表达式等模块的回归问题,并修复了CVE-2025-12084等安全漏洞。此版本标志着自由线程模式(移除GIL)正式获得官方支持,是Python发展的重要里程碑。
错误做法(在模板里转):
@app.route("/user/<uid>")
def get_user(uid):
user = db.users.find_one({"_id": ObjectId(uid)})
return render_template("user.html", user=user) # 模板里 {{ user._id }} 会崩</uid>
正确做法(在数据返回前统一处理):
- 用
json_util.dumps()(仅调试用,不推荐生产) - 更稳的方式:封装一个
serialize_doc(doc)函数,在所有查询后调用 - 示例:
from bson import ObjectId
from datetime import datetime
<p>def serialize_doc(doc):
if doc is None:
return None
for k, v in doc.items():
if isinstance(v, ObjectId):
doc[k] = str(v)
elif isinstance(v, datetime):
doc[k] = v.isoformat()
return doc</p><h1>使用</h1><p>user = serialize_doc(db.users.find_one({"_id": ObjectId(uid)}))</p>
注意:find() 返回的是游标,不是 dict,所以 serialize_doc(list(cursor)) 才对;find_one() 返回 dict,可直接传入。
集合不存在时不会报错,但 insert 不等于建库建表
MongoDB 的“懒创建”特性常被误解:执行 db.my_collection.insert_one({...}) 确实会自动创建集合,但不会验证字段合法性、不设索引、不加校验规则。
上线前必须手动初始化:
- 检查集合是否存在:
collection_name in db.list_collection_names() - 创建唯一索引(比如邮箱):
db.users.create_index("email", unique=True) - 设置 TTL 索引(比如 session):
db.sessions.create_index("expires_at", expireAfterSeconds=0) - 如果用了 MongoDB 3.6+,可定义 validator:
db.create_collection("users", validator={...})
没做这些,上线后才发现重复注册、session 泄露、数据格式混乱,修复成本远高于初始化时多写 3 行。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










