這一篇是“100天精通Python”系列的第 40 天主題非常直接用pymongo操作 MongoDB 數(shù)據(jù)庫從安裝、連接到增刪改查再到批量任務(wù)和索引優(yōu)化最后帶上接口封裝思路和常見問題排查。MongoDB 是目前最常用的文檔型 NoSQL 數(shù)據(jù)庫數(shù)據(jù)以 BSON 格式存儲業(yè)務(wù)上可以理解成“存在數(shù)據(jù)庫里的 JSON 對象”。Python 操作 MongoDB 最主流的方案就是官方驅(qū)動pymongo它的 API 設(shè)計(jì)比較簡潔和 MongoDB 的 Shell 語法很接近適合做爬蟲存儲、日志收集、用戶行為分析、配置中心這類場景。本文不繞彎子直接分成幾個部分先看 pymongo 的核心能力、再準(zhǔn)備環(huán)境、然后啟動 MongoDB 服務(wù)、接著用代碼把增刪改查和批量任務(wù)全部跑一遍最后給出索引優(yōu)化、接口 API 封裝和常見坑的排查方法。如果你正在學(xué) Python 操作數(shù)據(jù)庫或者準(zhǔn)備把項(xiàng)目里的數(shù)據(jù)存儲切到 MongoDB這篇文章可以直接作為操作手冊。1. 核心能力速覽能力項(xiàng)說明項(xiàng)目類型Python 官方 MongoDB 驅(qū)動程序核心功能連接 MongoDB、數(shù)據(jù)庫/集合管理、增刪改查、聚合管道、索引管理、批量寫入、事務(wù)支持支持平臺Windows、Linux、macOSPython 版本建議 Python 3.8 及以上具體以 pymongo 當(dāng)前版本要求為準(zhǔn)MongoDB 版本支持 MongoDB 3.6 及以上新版 pymongo 建議配合 MongoDB 4.2 使用安裝方式pip install pymongo通信方式TCP默認(rèn)端口 27017批量能力insert_many、bulk_write、游標(biāo)批量遍歷接口 API配合 Flask/FastAPI 可封裝成 HTTP 接口適合場景爬蟲數(shù)據(jù)存儲、日志收集、用戶行為記錄、配置中心、快速原型開發(fā)不適合場景復(fù)雜多表關(guān)聯(lián)查詢、強(qiáng)事務(wù)一致性、圖關(guān)系分析從材料看pymongo 的定位是“用 Python 的方式操作 MongoDB”所以你在 MongoDB Shell 里能做的操作在 pymongo 里基本都能對應(yīng)著寫出來。安裝成本很低沒有額外的編譯依賴只要有網(wǎng)絡(luò)環(huán)境一條命令就能裝好。2. 適用場景與使用邊界先判斷這個工具適不適合你。MongoDB 的文檔模型比關(guān)系型表結(jié)構(gòu)更靈活一個集合collection里的文檔不要求字段完全一致新增字段不需要先改表結(jié)構(gòu)。這種特性非常適合數(shù)據(jù)格式多變、字段可能長期演進(jìn)的業(yè)務(wù)比如爬蟲保存的頁面數(shù)據(jù)、埋點(diǎn)日志、用戶配置信息。pymongo 在這種場景下能把 Python 字典直接寫入數(shù)據(jù)庫讀出來也是字典和 Python 對象之間的轉(zhuǎn)換成本很低。但也要注意邊界。如果你需要大量JOIN操作、多行事務(wù)、復(fù)雜聚合統(tǒng)計(jì)MongoDB 雖然支持一部分但并不是最優(yōu)選擇。傳統(tǒng) ERP、金融賬務(wù)系統(tǒng)還是建議用 MySQL 或 PostgreSQL。pymongo 只是數(shù)據(jù)庫驅(qū)動它本身不會幫你解決數(shù)據(jù)建模問題集合設(shè)計(jì)不好同樣的查詢會明顯變慢。使用邊界方面重點(diǎn)提三點(diǎn)不要把 MongoDB 服務(wù)直接暴露到公網(wǎng)必須開啟訪問認(rèn)證否則很容易被掃描和寫入惡意數(shù)據(jù)。涉及用戶隱私、電話號碼、身份信息等敏感字段建議在寫入前加密或脫敏至少不要明文保存在沒有權(quán)限控制的集合里。要定期備份數(shù)據(jù)特別是使用副本集或單機(jī)部署時(shí)否則誤操作delete_many后恢復(fù)成本很高。3. 環(huán)境準(zhǔn)備與前置條件在寫代碼之前需要把 Python、MongoDB、pymongo 三樣?xùn)|西準(zhǔn)備好。3.1 安裝 MongoDBMongoDB 需要單獨(dú)安裝。Windows 用戶可以下載 MongoDB Community Server 的 MSI 安裝包安裝時(shí)選擇“Install MongoDB as a Service”可以注冊成 Windows 服務(wù)開機(jī)自啟。Linux 用戶可以通過 apt 或 yum 安裝macOS 用戶可以用 Homebrew。安裝完成后確認(rèn)mongod命令可用。Windows 下如果配置了服務(wù)可以直接啟動服務(wù)如果沒配置可以手動指定數(shù)據(jù)目錄啟動# Linux/macOS 示例需要先創(chuàng)建數(shù)據(jù)目錄 mkdir -p /data/db mongod --dbpath /data/db --port 27017Windows 命令行啟動方式mongod --dbpath D:\mongodb\data --port 27017啟動日志中出現(xiàn)Waiting for connections說明服務(wù)已經(jīng)正常監(jiān)聽 27017 端口。如果只想快速驗(yàn)證也可以不用本地安裝直接在測試服務(wù)器上用 Docker 啟動docker run -d --name mongodb -p 27017:27017 mongo:6.0mongo:6.0是示例鏡像標(biāo)簽具體版本可以根據(jù)自己的環(huán)境調(diào)整。3.2 安裝 Python 和 pymongoPython 建議用 3.8 以上版本。先確認(rèn) Python 環(huán)境python --version然后安裝 pymongopip install pymongo如果需要操作 MongoDB 的聚合管道、GridFS、加密等擴(kuò)展功能還可以安裝pymongo[srv]或pymongo[encryption]但基礎(chǔ)使用只需要一個pymongo包。驗(yàn)證安裝是否成功python -c import pymongo; print(pymongo.version)能輸出版本號說明驅(qū)動已經(jīng)可用。3.3 檢查 MongoDB 服務(wù)狀態(tài)在 Python 中連接之前建議先用命令行工具mongosh或者直接通過 pymongo 檢查。最簡單的檢查方式是寫一段連接代碼后面會講到。只要端口沒被防火墻攔截連接本身不會太復(fù)雜。4. 連接 MongoDB 數(shù)據(jù)庫連接 MongoDB 的核心類是MongoClient。這一步要解決兩件事如何連接本地服務(wù)、如何帶認(rèn)證信息連接。4.1 創(chuàng)建 MongoClient 連接先把代碼寫出來from pymongo import MongoClient # 本地默認(rèn)端口連接 client MongoClient(mongodb://127.0.0.1:27017/) # 獲取數(shù)據(jù)庫不存在會自動創(chuàng)建 db client.mydb # 獲取集合不存在會自動創(chuàng)建 collection db.users這里client.mydb和client[mydb]是等價(jià)的。MongoDB 是“懶創(chuàng)建”機(jī)制只有實(shí)際執(zhí)行寫入操作時(shí)數(shù)據(jù)庫和集合才會真正創(chuàng)建。4.2 帶用戶名密碼的連接如果 MongoDB 開啟了認(rèn)證就需要在連接串里帶上賬號client MongoClient( mongodb://admin:password123127.0.0.1:27017/?authSourceadmin )authSource表示認(rèn)證庫常見為admin。如果你的賬號建立在某個業(yè)務(wù)庫下authSource就寫對應(yīng)的庫名。生產(chǎn)環(huán)境建議用環(huán)境變量保存連接串不要硬編碼到代碼里。4.3 連接參數(shù)MongoClient還支持一些常用參數(shù)client MongoClient( mongodb://127.0.0.1:27017/, serverSelectionTimeoutMS5000, # 5秒內(nèi)選不到服務(wù)器就報(bào)錯 connectTimeoutMS3000, maxPoolSize50 )maxPoolSize控制連接池大小。默認(rèn)的MongoClient是懶連接真正發(fā)起操作時(shí)才會建立連接。如果你在腳本里只是創(chuàng)建 client 并不操作它不會立刻報(bào)錯。4.4 驗(yàn)證連接是否成功from pymongo import MongoClient client MongoClient(mongodb://127.0.0.1:27017/, serverSelectionTimeoutMS5000) # 向服務(wù)器發(fā)送 ping 命令 try: client.admin.command(ping) print(MongoDB 連接成功) except Exception as e: print(連接失敗:, e)這個ping命令是驗(yàn)證網(wǎng)絡(luò)和服務(wù)狀態(tài)最直接的方式。如果失敗先檢查服務(wù)有沒有啟動、端口是否正確。5. 文檔插入操作MongoDB 的基本數(shù)據(jù)單元是文檔在 pymongo 中對應(yīng) 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的列表。批量插入比循環(huán)insert_one效率高很多因?yàn)闇p少了客戶端和服務(wù)端的往返次數(shù)。5.3 插入數(shù)據(jù)后的驗(yàn)證count collection.count_documents({}) print(當(dāng)前用戶數(shù)量:, count)注意count()方法在新版 pymongo 中已經(jīng)棄用統(tǒng)一使用count_documents。6. 文檔查詢操作查詢是 pymongo 里用得最多的部分。掌握查詢語法基本就能解決 80% 的日常需求。6.1 查詢單條文檔find_oneuser collection.find_one({name: 張三}) print(user)返回的是字典找不到時(shí)返回None。和 MongoDB Shell 的findOne行為一致。6.2 查詢多條文檔findcursor collection.find({age: {$gte: 25}}) for user in cursor: print(user[name], user[age])find返回一個Cursor對象是一個可迭代的游標(biāo)?？梢杂胠ist()轉(zhuǎn)換成列表但如果數(shù)據(jù)量很大不建議一次性全部加載到內(nèi)存。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 排序、分頁、計(jì)數(shù)# 按年齡降序排序 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是傳統(tǒng)分頁方案。數(shù)據(jù)量超過幾萬頁時(shí)更推薦用_id或排序字段做游標(biāo)分頁避免大偏移量導(dǎo)致性能下降。6.5 正則和數(shù)組查詢# 名字以 張 開頭 users collection.find({name: {$regex: ^張}}) # tags 數(shù)組中包含 python users collection.find({tags: python}) # tags 數(shù)組同時(shí)包含 python 和 mongodb users collection.find({tags: {$all: [python, mongodb]}})數(shù)組查詢在標(biāo)簽系統(tǒng)、權(quán)限系統(tǒng)里很常用$all表示數(shù)組字段必須同時(shí)包含多個值。7. 文檔更新操作更新操作最常用的是update_one和update_many。更新時(shí)要注意直接傳入整個新文檔會替換原文檔更新指定字段必須使用更新操作符比如$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表示匹配到的文檔數(shù)modified_count表示實(shí)際被修改的文檔數(shù)。這里能發(fā)現(xiàn)如果新值和舊值一樣modified_count是 0。7.2 批量更新update_many# 將所有北京用戶的年齡加 1 result collection.update_many( {city: 北京}, {$inc: {age: 1}} ) print(result.modified_count)$inc是數(shù)值增減操作適合計(jì)數(shù)、積分、庫存等場景。7.3 數(shù)組更新$push和$addToSet# 向 tags 數(shù)組追加一個值 collection.update_one( {name: 張三}, {$push: {tags: pymongo}} ) # 如果值已經(jīng)存在則不重復(fù)添加 collection.update_one( {name: 張三}, {$addToSet: {tags: pymongo}} )$push每次都會追加$addToSet會去重。需要維護(hù)標(biāo)簽、關(guān)注列表、操作日志時(shí)優(yōu)先考慮$addToSet。7.4 替換整條文檔replace_onenew_doc { name: 張三, age: 30, city: 廣州, tags: [python] } collection.replace_one({name: 張三}, new_doc)replace_one會刪除原文檔中其他字段用新文檔整體替換。除非業(yè)務(wù)明確需要否則更推薦$set做局部更新避免誤刪字段。7.5 upsert 更新或插入result collection.update_one( {name: 孫七}, {$set: {age: 18, city: 杭州}}, upsertTrue ) print(result.upserted_id)加上upsertTrue后如果查詢條件匹配不到文檔就會插入一條新文檔。這個能力在“存在則更新不存在則創(chuàng)建”的業(yè)務(wù)中非常實(shí)用。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()清除整個數(shù)據(jù)庫client.drop_database(mydb)刪除是不可逆操作建議在執(zhí)行delete_many前先用count_documents看一下確定影響范圍或者在測試庫上先驗(yàn)證。9. 索引與批量任務(wù)索引是 MongoDB 查詢性能的核心。沒有索引時(shí)MongoDB 必須做全表掃描數(shù)據(jù)到幾十萬條后查詢延遲會明顯上升。9.1 創(chuàng)建索引# 單字段索引 collection.create_index(name) # 復(fù)合索引 collection.create_index([(city, 1), (age, -1)]) # 唯一索引 collection.create_index(email, uniqueTrue) # 查看集合索引 for index in collection.list_indexes(): print(index)1表示升序-1表示降序。如果業(yè)務(wù)查詢經(jīng)常同時(shí)按city和age過濾排序適合建復(fù)合索引。唯一索引可以用來防止郵箱、手機(jī)號等字段重復(fù)。9.2 批量寫入與bulk_write批量任務(wù)指的是在一次請求里執(zhí)行多條寫操作。前面提到的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適合做數(shù)據(jù)同步、清洗、ETL 任務(wù)。它默認(rèn)按順序執(zhí)行如果中間某條失敗后面的操作可能不會繼續(xù)可以使用orderedFalse讓 MongoDB 繼續(xù)執(zhí)行剩余操作result collection.bulk_write(operations, orderedFalse)9.3 批量查詢與游標(biāo)遍歷如果要對全量數(shù)據(jù)做處理不要一次性list()出來應(yīng)該使用游標(biāo)分塊遍歷batch_size 500 cursor collection.find({}).batch_size(batch_size) for doc in cursor: # 處理每條文檔 process(doc)batch_size控制每次從服務(wù)端取回的文檔數(shù)量可以避免把大量數(shù)據(jù)一次性加載到內(nèi)存。10. 接口 API 調(diào)用示例pymongo 本身不是 HTTP 服務(wù)但實(shí)際項(xiàng)目中經(jīng)常需要把數(shù)據(jù)庫操作封裝成接口讓前端或內(nèi)部服務(wù)調(diào)用。下面用 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 轉(zhuǎn)成字符串方便 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)這是一個通用示例實(shí)際部署時(shí)要注意接口必須做權(quán)限校驗(yàn)不能把數(shù)據(jù)庫接口直接裸露給公網(wǎng)。入?yún)⒁ｒ?yàn)避免用戶傳入$set之類的內(nèi)容修改非預(yù)期字段。ObjectId轉(zhuǎn)換失敗時(shí)要做異常處理。大規(guī)模接口服務(wù)建議使用 FastAPI Pydantic做自動化參數(shù)校驗(yàn)和文檔輸出。也可以用curl快速驗(yàn)證接口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 是內(nèi)存友好的數(shù)據(jù)庫嗎實(shí)際上它依賴操作系統(tǒng)緩存索引和熱數(shù)據(jù)盡可能放在內(nèi)存中查詢速度很快。單機(jī)部署時(shí)MongoDB 進(jìn)程會占用較多內(nèi)存這是正?，F(xiàn)象因?yàn)?LRU 緩存會主動利用空閑內(nèi)存。觀察資源占用時(shí)重點(diǎn)看以下指標(biāo)觀察項(xiàng)方法說明MongoDB 內(nèi)存占用任務(wù)管理器或mongostat緩存大小會動態(tài)變化連接數(shù)db.serverStatus().connections連接池如果耗盡請求會阻塞查詢耗時(shí)explain(executionStats)檢查是否走索引磁盤 IOiostat或云監(jiān)控寫入量大時(shí)磁盤容易成為瓶頸Python 端連接池maxPoolSize高并發(fā)場景要合理設(shè)置explain是排查慢查詢的重要工具cursor collection.find({city: 北京}).sort(age, -1) explain_result cursor.explain() print(explain_result[queryPlanner][winningPlan])如果 winning plan 顯示COLLSCAN說明沒有走索引。這時(shí)應(yīng)該考慮建立合適的索引。使用 pymongo 時(shí)有幾個性能習(xí)慣一個進(jìn)程內(nèi)復(fù)用同一個MongoClient不要每次操作都新建連接。批量插入優(yōu)先insert_many不要循環(huán)insert_one。查詢時(shí)只返回需要的字段減少網(wǎng)絡(luò)傳輸。避免一次find后把所有結(jié)果轉(zhuǎn)成 list改用游標(biāo)。寫入量大的業(yè)務(wù)開啟確認(rèn)寫w1即可不要用wmajority做不必要的確認(rèn)除非你確實(shí)需要強(qiáng)一致。12. 常見問題與排查方法問題現(xiàn)象可能原因排查方式解決方案連接超時(shí)ServerSelectionTimeoutErrorMongoDB 服務(wù)未啟動、端口錯誤、防火墻攔截檢查mongod進(jìn)程、netstat -ano | findstr 27017、Telnet 端口啟動服務(wù)確認(rèn)連接串端口放行防火墻認(rèn)證失敗Authentication failed用戶名密碼錯誤、認(rèn)證庫不對、用戶沒有權(quán)限用mongosh測試認(rèn)證信息確認(rèn)authSource指向正確的認(rèn)證庫重新設(shè)置用戶插入文檔報(bào)DocumentTooLarge單條文檔超過 16 MB 限制查看報(bào)錯內(nèi)容壓縮字段、拆分文檔或使用 GridFS 存儲大文件count()報(bào)錯或告警新版 pymongo 移除了舊count查看版本變更改用count_documents({})查詢速度很慢沒有索引或索引不合理執(zhí)行explain查看COLLSCAN創(chuàng)建單字段或復(fù)合索引批量寫入部分失敗orderedTrue模式下中途出錯檢查報(bào)錯開啟日志使用orderedFalse或?qū)﹀e誤記錄重試連接數(shù)過多每次操作創(chuàng)建新MongoClient且未復(fù)用查看連接數(shù)復(fù)用全局 client調(diào)整maxPoolSize數(shù)據(jù)亂碼字符集問題檢查寫入編碼客戶端統(tǒng)一使用 UTF-8刪除后數(shù)據(jù)無法恢復(fù)沒有備份檢查備份策略使用mongodump定期備份部署副本集最常見的問題是“明明本地裝了 MongoDB但 Python 連不上”絕大多數(shù)原因是 MongoDB 服務(wù)沒有啟動或者連接串端口寫錯。建議先跑一下ping命令再排查其他環(huán)節(jié)。13. 最佳實(shí)踐與使用建議pymongo 的 API 雖然簡單但項(xiàng)目落地時(shí)還是有一些工程化經(jīng)驗(yàn)可以提前規(guī)避問題。第一連接管理要統(tǒng)一。建議在項(xiàng)目入口創(chuàng)建全局MongoClient通過配置模塊讀取連接串不要在每個函數(shù)里都MongoClient(...)。這樣既能復(fù)用連接池也方便切測試庫和生產(chǎn)庫。# 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第二集合名和字段名要有規(guī)范。集合名建議使用復(fù)數(shù)名詞比如users、orders。字段名統(tǒng)一使用小寫和下劃線避免出現(xiàn)userId和user_id混用。字段名風(fēng)格不統(tǒng)一后期寫查詢條件時(shí)會非常痛苦。第三建立索引要有計(jì)劃。不要上來就給所有字段加索引索引會占用磁盤空間并且降低寫入性能。應(yīng)該先分析查詢條件對高頻過濾字段建立索引對排序字段建立復(fù)合索引。第四生產(chǎn)環(huán)境一定要開啟認(rèn)證和訪問控制。MongoDB 默認(rèn)是沒有賬號密碼的如果監(jiān)聽在0.0.0.0相當(dāng)于把數(shù)據(jù)裸奔在公網(wǎng)上。至少要做兩件事服務(wù)監(jiān)聽內(nèi)網(wǎng) IP創(chuàng)建專用業(yè)務(wù)賬號并分配最小權(quán)限。第五涉及敏感數(shù)據(jù)時(shí)要加密和脫敏。不能把email、phone、id_card這類字段明文存儲除非數(shù)據(jù)庫網(wǎng)絡(luò)和權(quán)限完全隔離。建議寫入前用加密算法處理展示時(shí)再解密或者完成業(yè)務(wù)后自動清理。第六批量任務(wù)要加上日志和失敗重試機(jī)制。用bulk_write處理大量數(shù)據(jù)時(shí)如果中間出錯最好捕獲異常并記錄失敗的文檔 key后續(xù)單獨(dú)重試。不要在一次任務(wù)里寫入上萬條還不打印任何日志出了問題很難定位。第七備份策略要提前設(shè)計(jì)。至少使用mongodump做定時(shí)全量備份有條件的部署副本集實(shí)現(xiàn)高可用。對于核心業(yè)務(wù)還可以開啟 op log用增量備份降低數(shù)據(jù)丟失風(fēng)險(xiǎn)。14. 總結(jié)與下一步這一篇把 Python 操作 MongoDB 的完整路徑過了一遍環(huán)境準(zhǔn)備、服務(wù)啟動、pymongo 連接、插入、查詢、更新、刪除、索引、批量任務(wù)、接口封裝和排錯方法。最值得先動手驗(yàn)證的是insert_onefind_oneupdate_onedelete_one這條基礎(chǔ)鏈路跑通之后MongoDB 的核心用法你就已經(jīng)掌握了。最容易踩的坑有三個第一MongoDB 服務(wù)沒啟動就急著跑 Python 代碼結(jié)果報(bào)連接超時(shí)第二更新文檔時(shí)忘記用$set結(jié)果整條文檔被覆蓋第三大量數(shù)據(jù)查詢時(shí)沒有建索引數(shù)據(jù)量上去之后接口變慢。如果這周有時(shí)間建議繼續(xù)往下練幾個方向MongoDB 聚合管道aggregate做分組統(tǒng)計(jì)GridFS存大文件副本集事務(wù)操作以及motor異步驅(qū)動對接 FastAPI。等基礎(chǔ)增刪改查熟練之后再去看聚合和事務(wù)會輕松很多。建議把這篇文章收藏備用下次需要操作 MongoDB 時(shí)可以直接對照著寫代碼。