前言先把最容易搞错的一点说清楚MongoDB 不是关系型数据库。它没有 SQL、没有表、没有「行」和「列」、没有 JOIN、没有固定表结构。它是文档型数据库数据以 BSON 文档一种类似 JSON 的二进制格式存储一层层嵌套同一个集合里的文档字段可以互不相同。所以「用 Python 操作 MongoDB」和「用 Python 操作 MySQL」在思维方式上是两套东西前者是「把整个对象存进去、按条件捞出来」后者是「把对象拆成列、用 JOIN 拼回去」。把关系型的思路硬搬到 MongoDB 上通常会得到一堆嵌套很深的文档和一堆在应用层手写的「JOIN」。第二个常见误解「MongoDB 不用设计 schema」——不是不用设计是 schema从数据库层挪到了应用层。数据库不再拦你写错字段所以校验、约定、索引都得你自己在代码里管。少了约束不等于少了责任。本文用官方驱动PyMongo当前主版本 4.x走一遍完整流程连接、增删改查、索引、以及最容易被忽略的输入校验。所有来自外部的数据在进入查询条件前都做类型校验。一、安装与连接python -m pip install pymongo# 适用于 Python 3.9PyMongo 4.ximport osfrom pymongo import MongoClient# 推荐用连接串形式凭据从环境变量读uri os.environ.get(MONGO_URI, mongodb://127.0.0.1:27017)client MongoClient(uri, serverSelectionTimeoutMS3000)try:info client.server_info() # 触发一次真实往返确认能连上print(MongoDB 版本:, info[version])print(已有数据库:, client.list_database_names())finally:client.close()要点serverSelectionTimeoutMS建议设小一点。默认值较长服务没起来时程序会「卡住」很久才报错看起来像死机。MongoClient的连接是惰性建立的。构造它不会立刻连服务器第一次真正操作时才连。所以「连不上」的错误往往在第一次查询时才冒出来而不是在构造那一行。想尽早发现就主动调一次client.server_info()。MongoClient是线程安全的应该全进程复用一个。它内部维护连接池每次操作都新建一个 client 会把连接耗尽。client.close()显式关闭尤其是脚本结束时。常驻服务里则在退出钩子里关一次。二、数据库与集合用到才创建# 适用于 Python 3.9db client[blog] # 不存在也不会报错只是拿到一个引用posts db[posts] # 集合同理# 空库/空集合不会出现在 list_database_names() 里# 直到你写入第一条数据print(db.list_collection_names())关键行为MongoDB 的数据库和集合是「首次写入时才真正创建」的。你client[blog]只是拿了个引用服务器上什么都没有。所以想确认集合是否真的存在不要靠「取了引用就算建好了」用db.list_collection_names()看。字段的「类型」也是首次写入时定下来的更准确地说是每次写入的实际值决定的。同名字段在不同文档里类型不同MongoDB 不会拦你——但你的查询和索引会因此变得不可预测。三、插入文档# 适用于 Python 3.9from datetime import datetime, timezoneresult posts.insert_one({title: MongoDB 入门,author: 张三,tags: [数据库, NoSQL],views: 0,created_at: datetime.now(timezone.utc),})print(插入的 _id , result.inserted_id) # 未指定时服务器生成 ObjectIdresult posts.insert_many([{title: 索引怎么建, author: 李四, tags: [性能], views: 0},{title: 聚合管道, author: 张三, tags: [数据库], views: 0},], orderedFalse)print(插入的 _id 列表 , result.inserted_ids)要点_id是每个文档的主键默认由服务器生成ObjectId。可以自己指定但必须在整个集合内唯一。insert_one通过result.inserted_id拿回来。insert_many(..., orderedFalse)表示「遇到错误继续插入剩下的」。默认orderedTrue一旦某条失败后面的都不插。批量导入时用orderedFalse能一次看到所有冲突代价是失败的具体是哪些需要从异常里逐条读。datetime要带时区。用datetime.now(timezone.utc)而不是datetime.now()不带时区的时间存进去以后做时间区间查询会非常痛苦。没有「先建集合再插」这一步。集合不存在时插入会自动创建。四、查询# 适用于 Python 3.9# 1) 查一条没有匹配时返回 None不抛异常doc posts.find_one({author: 张三})if doc is not None:print(doc[title])# 2) 查多条返回游标可以链式加排序和分页cursor (posts.find({tags: 数据库}, {title: 1, author: 1}).sort(created_at, -1).skip(0).limit(10))for d in cursor:print(d[_id], d[title])# 3) 计数n posts.count_documents({author: 张三})print(张三的文章数:, n)要点find_one()查不到时返回None不抛异常。这是和「按主键取记录」类 API 最大的差别必须判空否则下一行doc[title]直接TypeError。find()的第二个参数是投影projection{title: 1, author: 1}表示只返回这两个字段_id默认仍会返回除非显式写_id: 0。只取需要的字段能显著减少网络传输。游标是惰性的find()不会立刻把数据全拉回来遍历时才分批取。所以find()后面可以链式.sort()/.skip()/.limit()。查询嵌套字段用点号路径{author.name: 张三}查数组元素直接匹配数组里的值{tags: 数据库}。五、更新用操作符不要整份替换# 适用于 Python 3.9result posts.update_one({title: MongoDB 入门},{$inc: {views: 1}, $set: {updated_at: datetime.now(timezone.utc)}},)print(匹配:, result.matched_count, 实际修改:, result.modified_count)要点update_one(filter, update)的第二个参数必须是「更新操作符文档」比如$set设置字段、$inc数值自增、$push往数组追加、$unset删除字段。如果第二个参数里没有$开头的操作符PyMongo 会把它当成「整份替换文档」——这是最常见的用法错误会导致其他字段全部消失。matched_count和modified_count含义不同前者是「匹配到几条」后者是「实际改动了几条」。用同样的值再$set一次matched_count是 1modified_count是 0。要「整份替换」就用replace_one(filter, replacement)语义明确得多。upsertTrue表示「没有匹配就插入一条」。注意 upsert 走的路径和普通更新不同如果更新里有$setOnInsert这类只在插入时生效的操作符要单独确认行为。六、删除与索引# 适用于 Python 3.9r posts.delete_one({title: 聚合管道})print(删除条数:, r.deleted_count) # 字段名是 deleted_count# 建唯一索引从数据库层保证业务唯一性posts.create_index(title, uniqueTrue)要点删除结果的字段是deleted_count不是deleted_counts。唯一性约束在 MongoDB 里靠唯一索引实现没有别的「唯一键」概念。有了唯一索引重复插入会抛pymongo.errors.DuplicateKeyError# 适用于 Python 3.9import pymongotry:posts.insert_one({title: MongoDB 入门, author: 张三})except pymongo.errors.DuplicateKeyError:print(标题已存在忽略这次插入)pymongo.errors.PyMongoError是所有 PyMongo 异常的基类。要「捕获所有数据库异常」捕它就够要区分类型DuplicateKeyError唯一键冲突、ConnectionFailure连不上、OperationFailure服务端操作失败各有各的用途。七、把校验做在前面防止查询条件被「变形」这是 MongoDB 应用里最需要警惕的一类安全问题。关系型数据库用参数化查询把「值」和「语法」彻底分开。MongoDB 的查询条件是 Python 字典字典本身可以被构造成「带查询操作符的文档」。如果代码把外部传入的整个字典直接当查询条件用那么攻击者传进来的就不只是「一个值」而是一段查询逻辑——「等于 X」可以被变成「不等于 X」「存在即匹配」之类的完全不同的语义从而绕过本应起作用的校验。防御手段只有一条核心原则进入查询条件的值必须是确定的类型。# 适用于 Python 3.9def find_post_by_title(posts, title):只接受字符串标题其他类型一律拒绝。if not isinstance(title, str):raise ValueError(title 必须是字符串)title title.strip()if not title or len(title) 200:raise ValueError(title 长度不合法)return posts.find_one({title: title})def find_posts_by_author(posts, author, limit20):if not isinstance(author, str):raise ValueError(author 必须是字符串)if not isinstance(limit, int) or not (1 limit 100):raise ValueError(limit 必须是 1~100 的整数)# 把值放进操作符里保证是「相等匹配」的语义return list(posts.find({author: {$eq: author.strip()}}).limit(limit))配套的做法用类型注解 显式isinstance校验把「必须是什么类型」写死在函数入口。显式写出$eq当值被明确包在{$eq: value}里时value不可能再被解释成别的东西。不要接受「整个查询条件来自外部」。接口层只接收具体字段标题、作者、分页参数由服务端代码自己组装查询。应用账号最小权限。给 MongoDB 用户只授予需要的那几个库/集合的读写权限不要用管理账号跑业务。连接必须认证不要把 MongoDB 直接暴露在公网。常见坑点1. 用关系型的思路设计文档❌ 把「文章」和「评论」拆成两个集合然后在 Python 里循环查询拼成评论列表——这等于手写 JOIN。✅ 根据访问方式设计一起读的、一起写的、数量有界的数据放在同一个文档里比如评论内嵌在文章里数量无界的比如一个热门帖的十万条评论才拆出去单独存。2.find_one()返回None时直接取字段❌doc posts.find_one({title: t}); print(doc[author])——查不到时抛TypeError。✅if doc is None: ...先判空或者用(posts.find_one(...) or {}).get(author)。3. 更新时忘了写操作符❌posts.update_one({title: t}, {views: 100})——第二个参数没有$被当成整份替换其他字段全部丢失。✅posts.update_one({title: t}, {$set: {views: 100}})。4. 把外部字典直接当查询条件❌posts.find_one(request_data)——传入的字典可以携带查询操作符把「相等匹配」变成完全不同的语义绕过校验。✅ 只接收具体字段并做类型校验服务端自己组装条件if not isinstance(name, str):raise ValueError(name 必须是字符串)posts.find_one({name: {$eq: name}})5. 用matched_count判断「值有没有变」❌ 看到matched_count 1就认为数据更新了——其实可能新旧值相同什么都没改。✅ 判断是否真有改动看modified_count判断记录是否存在看matched_count。6. 时间不带时区❌datetime.now()存进文档之后跨时区做范围查询结果差几个小时还找不到原因。✅ 统一用datetime.now(timezone.utc)展示时再按用户时区转换。7. 每次操作都新建MongoClient❌ 在函数内部MongoClient(...)用完不关——连接池不断新建很快耗尽。✅ 全进程一个MongoClient各线程共用退出时close()一次。8. 依赖「数据库会帮我保证唯一性」却没建索引❌ 用「先查再插」保证标题唯一两个请求并发时同时通过检查插出两条。✅ 建唯一索引posts.create_index(title, uniqueTrue)在应用层捕获DuplicateKeyError给出友好提示。总结环节正确做法定位文档型数据库无 SQL / 表 / JOINschema 约束在应用层连接全进程复用一个MongoClient设短serverSelectionTimeoutMS显式close()建库表首次写入才真正创建集合不存在时插入会自动建插入insert_one/insert_many(orderedFalse)用带时区的datetime查询find_one返回None要判空投影只取需要字段游标惰性、可链式更新必须用$set/$inc等操作符否则会被当成整份替换唯一性靠唯一索引捕获DuplicateKeyError安全查询条件只接受确定类型的具体字段显式用$eq账号最小权限MongoDB 用起来比关系型数据库「松」但松不等于省事约束不会消失只会转移到你的代码里。把类型校验和索引这两件事做扎实就能既享受文档模型的灵活又不至于在数据一致性上翻车。
前言先把最容易搞错的一点说清楚MongoDB 不是关系型数据库。它没有 SQL、没有表、没有「行」和「列」、没有 JOIN、没有固定表结构。它是文档型数据库数据以 BSON 文档一种类似 JSON 的二进制格式存储一层层嵌套同一个集合里的文档字段可以互不相同。所以「用 Python 操作 MongoDB」和「用 Python 操作 MySQL」在思维方式上是两套东西前者是「把整个对象存进去、按条件捞出来」后者是「把对象拆成列、用 JOIN 拼回去」。把关系型的思路硬搬到 MongoDB 上通常会得到一堆嵌套很深的文档和一堆在应用层手写的「JOIN」。第二个常见误解「MongoDB 不用设计 schema」——不是不用设计是 schema从数据库层挪到了应用层。数据库不再拦你写错字段所以校验、约定、索引都得你自己在代码里管。少了约束不等于少了责任。本文用官方驱动PyMongo当前主版本 4.x走一遍完整流程连接、增删改查、索引、以及最容易被忽略的输入校验。所有来自外部的数据在进入查询条件前都做类型校验。一、安装与连接python -m pip install pymongo# 适用于 Python 3.9PyMongo 4.ximport osfrom pymongo import MongoClient# 推荐用连接串形式凭据从环境变量读uri os.environ.get(MONGO_URI, mongodb://127.0.0.1:27017)client MongoClient(uri, serverSelectionTimeoutMS3000)try:info client.server_info() # 触发一次真实往返确认能连上print(MongoDB 版本:, info[version])print(已有数据库:, client.list_database_names())finally:client.close()要点serverSelectionTimeoutMS建议设小一点。默认值较长服务没起来时程序会「卡住」很久才报错看起来像死机。MongoClient的连接是惰性建立的。构造它不会立刻连服务器第一次真正操作时才连。所以「连不上」的错误往往在第一次查询时才冒出来而不是在构造那一行。想尽早发现就主动调一次client.server_info()。MongoClient是线程安全的应该全进程复用一个。它内部维护连接池每次操作都新建一个 client 会把连接耗尽。client.close()显式关闭尤其是脚本结束时。常驻服务里则在退出钩子里关一次。二、数据库与集合用到才创建# 适用于 Python 3.9db client[blog] # 不存在也不会报错只是拿到一个引用posts db[posts] # 集合同理# 空库/空集合不会出现在 list_database_names() 里# 直到你写入第一条数据print(db.list_collection_names())关键行为MongoDB 的数据库和集合是「首次写入时才真正创建」的。你client[blog]只是拿了个引用服务器上什么都没有。所以想确认集合是否真的存在不要靠「取了引用就算建好了」用db.list_collection_names()看。字段的「类型」也是首次写入时定下来的更准确地说是每次写入的实际值决定的。同名字段在不同文档里类型不同MongoDB 不会拦你——但你的查询和索引会因此变得不可预测。三、插入文档# 适用于 Python 3.9from datetime import datetime, timezoneresult posts.insert_one({title: MongoDB 入门,author: 张三,tags: [数据库, NoSQL],views: 0,created_at: datetime.now(timezone.utc),})print(插入的 _id , result.inserted_id) # 未指定时服务器生成 ObjectIdresult posts.insert_many([{title: 索引怎么建, author: 李四, tags: [性能], views: 0},{title: 聚合管道, author: 张三, tags: [数据库], views: 0},], orderedFalse)print(插入的 _id 列表 , result.inserted_ids)要点_id是每个文档的主键默认由服务器生成ObjectId。可以自己指定但必须在整个集合内唯一。insert_one通过result.inserted_id拿回来。insert_many(..., orderedFalse)表示「遇到错误继续插入剩下的」。默认orderedTrue一旦某条失败后面的都不插。批量导入时用orderedFalse能一次看到所有冲突代价是失败的具体是哪些需要从异常里逐条读。datetime要带时区。用datetime.now(timezone.utc)而不是datetime.now()不带时区的时间存进去以后做时间区间查询会非常痛苦。没有「先建集合再插」这一步。集合不存在时插入会自动创建。四、查询# 适用于 Python 3.9# 1) 查一条没有匹配时返回 None不抛异常doc posts.find_one({author: 张三})if doc is not None:print(doc[title])# 2) 查多条返回游标可以链式加排序和分页cursor (posts.find({tags: 数据库}, {title: 1, author: 1}).sort(created_at, -1).skip(0).limit(10))for d in cursor:print(d[_id], d[title])# 3) 计数n posts.count_documents({author: 张三})print(张三的文章数:, n)要点find_one()查不到时返回None不抛异常。这是和「按主键取记录」类 API 最大的差别必须判空否则下一行doc[title]直接TypeError。find()的第二个参数是投影projection{title: 1, author: 1}表示只返回这两个字段_id默认仍会返回除非显式写_id: 0。只取需要的字段能显著减少网络传输。游标是惰性的find()不会立刻把数据全拉回来遍历时才分批取。所以find()后面可以链式.sort()/.skip()/.limit()。查询嵌套字段用点号路径{author.name: 张三}查数组元素直接匹配数组里的值{tags: 数据库}。五、更新用操作符不要整份替换# 适用于 Python 3.9result posts.update_one({title: MongoDB 入门},{$inc: {views: 1}, $set: {updated_at: datetime.now(timezone.utc)}},)print(匹配:, result.matched_count, 实际修改:, result.modified_count)要点update_one(filter, update)的第二个参数必须是「更新操作符文档」比如$set设置字段、$inc数值自增、$push往数组追加、$unset删除字段。如果第二个参数里没有$开头的操作符PyMongo 会把它当成「整份替换文档」——这是最常见的用法错误会导致其他字段全部消失。matched_count和modified_count含义不同前者是「匹配到几条」后者是「实际改动了几条」。用同样的值再$set一次matched_count是 1modified_count是 0。要「整份替换」就用replace_one(filter, replacement)语义明确得多。upsertTrue表示「没有匹配就插入一条」。注意 upsert 走的路径和普通更新不同如果更新里有$setOnInsert这类只在插入时生效的操作符要单独确认行为。六、删除与索引# 适用于 Python 3.9r posts.delete_one({title: 聚合管道})print(删除条数:, r.deleted_count) # 字段名是 deleted_count# 建唯一索引从数据库层保证业务唯一性posts.create_index(title, uniqueTrue)要点删除结果的字段是deleted_count不是deleted_counts。唯一性约束在 MongoDB 里靠唯一索引实现没有别的「唯一键」概念。有了唯一索引重复插入会抛pymongo.errors.DuplicateKeyError# 适用于 Python 3.9import pymongotry:posts.insert_one({title: MongoDB 入门, author: 张三})except pymongo.errors.DuplicateKeyError:print(标题已存在忽略这次插入)pymongo.errors.PyMongoError是所有 PyMongo 异常的基类。要「捕获所有数据库异常」捕它就够要区分类型DuplicateKeyError唯一键冲突、ConnectionFailure连不上、OperationFailure服务端操作失败各有各的用途。七、把校验做在前面防止查询条件被「变形」这是 MongoDB 应用里最需要警惕的一类安全问题。关系型数据库用参数化查询把「值」和「语法」彻底分开。MongoDB 的查询条件是 Python 字典字典本身可以被构造成「带查询操作符的文档」。如果代码把外部传入的整个字典直接当查询条件用那么攻击者传进来的就不只是「一个值」而是一段查询逻辑——「等于 X」可以被变成「不等于 X」「存在即匹配」之类的完全不同的语义从而绕过本应起作用的校验。防御手段只有一条核心原则进入查询条件的值必须是确定的类型。# 适用于 Python 3.9def find_post_by_title(posts, title):只接受字符串标题其他类型一律拒绝。if not isinstance(title, str):raise ValueError(title 必须是字符串)title title.strip()if not title or len(title) 200:raise ValueError(title 长度不合法)return posts.find_one({title: title})def find_posts_by_author(posts, author, limit20):if not isinstance(author, str):raise ValueError(author 必须是字符串)if not isinstance(limit, int) or not (1 limit 100):raise ValueError(limit 必须是 1~100 的整数)# 把值放进操作符里保证是「相等匹配」的语义return list(posts.find({author: {$eq: author.strip()}}).limit(limit))配套的做法用类型注解 显式isinstance校验把「必须是什么类型」写死在函数入口。显式写出$eq当值被明确包在{$eq: value}里时value不可能再被解释成别的东西。不要接受「整个查询条件来自外部」。接口层只接收具体字段标题、作者、分页参数由服务端代码自己组装查询。应用账号最小权限。给 MongoDB 用户只授予需要的那几个库/集合的读写权限不要用管理账号跑业务。连接必须认证不要把 MongoDB 直接暴露在公网。常见坑点1. 用关系型的思路设计文档❌ 把「文章」和「评论」拆成两个集合然后在 Python 里循环查询拼成评论列表——这等于手写 JOIN。✅ 根据访问方式设计一起读的、一起写的、数量有界的数据放在同一个文档里比如评论内嵌在文章里数量无界的比如一个热门帖的十万条评论才拆出去单独存。2.find_one()返回None时直接取字段❌doc posts.find_one({title: t}); print(doc[author])——查不到时抛TypeError。✅if doc is None: ...先判空或者用(posts.find_one(...) or {}).get(author)。3. 更新时忘了写操作符❌posts.update_one({title: t}, {views: 100})——第二个参数没有$被当成整份替换其他字段全部丢失。✅posts.update_one({title: t}, {$set: {views: 100}})。4. 把外部字典直接当查询条件❌posts.find_one(request_data)——传入的字典可以携带查询操作符把「相等匹配」变成完全不同的语义绕过校验。✅ 只接收具体字段并做类型校验服务端自己组装条件if not isinstance(name, str):raise ValueError(name 必须是字符串)posts.find_one({name: {$eq: name}})5. 用matched_count判断「值有没有变」❌ 看到matched_count 1就认为数据更新了——其实可能新旧值相同什么都没改。✅ 判断是否真有改动看modified_count判断记录是否存在看matched_count。6. 时间不带时区❌datetime.now()存进文档之后跨时区做范围查询结果差几个小时还找不到原因。✅ 统一用datetime.now(timezone.utc)展示时再按用户时区转换。7. 每次操作都新建MongoClient❌ 在函数内部MongoClient(...)用完不关——连接池不断新建很快耗尽。✅ 全进程一个MongoClient各线程共用退出时close()一次。8. 依赖「数据库会帮我保证唯一性」却没建索引❌ 用「先查再插」保证标题唯一两个请求并发时同时通过检查插出两条。✅ 建唯一索引posts.create_index(title, uniqueTrue)在应用层捕获DuplicateKeyError给出友好提示。总结环节正确做法定位文档型数据库无 SQL / 表 / JOINschema 约束在应用层连接全进程复用一个MongoClient设短serverSelectionTimeoutMS显式close()建库表首次写入才真正创建集合不存在时插入会自动建插入insert_one/insert_many(orderedFalse)用带时区的datetime查询find_one返回None要判空投影只取需要字段游标惰性、可链式更新必须用$set/$inc等操作符否则会被当成整份替换唯一性靠唯一索引捕获DuplicateKeyError安全查询条件只接受确定类型的具体字段显式用$eq账号最小权限MongoDB 用起来比关系型数据库「松」但松不等于省事约束不会消失只会转移到你的代码里。把类型校验和索引这两件事做扎实就能既享受文档模型的灵活又不至于在数据一致性上翻车。
