引言：为什么选择Python+MongoDB？
在数据驱动的时代，开发者需要高效处理非结构化数据的能力。MongoDB作为文档型数据库的代表，与Python的简洁语法形成天然互补。本文将通过15个实战片段，带你从环境搭建到高级操作，掌握这对黄金组合的核心玩法。
环境准备：三步搭建开发环境
- 安装MongoDB社区版
Ubuntu/Debian
sudo apt-get install -y mongodb-org
macOS（使用Homebrew）
brew tap mongodb/brew
brew install mongodb-community@6.0
创建虚拟环境并安装PyMongo
python -m venv mongo_env
source mongo_env/bin/activate # Linux/macOS
pip install pymongo==4.6.1启动MongoDB服务
默认配置文件路径：/etc/mongod.conf
sudo systemctl start mongod # Linux系统
mongod --config /usr/local/etc/mongod.conf # macOS
连接管理：建立安全通信通道
基础连接示例
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['ecommerce'] # 创建/获取数据库
collection = db['products'] # 创建/获取集合
连接池配置（生产环境必备）
client = MongoClient(
'mongodb://user:pass@host1:27017,host2:27018/?replicaSet=myReplica',
maxPoolSize=50,
waitQueueTimeoutMS=2000
)
上下文管理器最佳实践
with MongoClient('mongodb://localhost:27017/') as client:
db = client.get_database('analytics')
# 在此作用域内执行操作
CRUD核心操作：数据处理的四把利剑
创建数据（Create）
插入单条文档
product = {
"name": "Wireless Mouse",
"price": 29.99,
"specs": {
"dpi": 8000,
"battery": "AAA x2"
},
"tags": ["electronics", "office"]
}
insert_result = collection.insert_one(product)
print(f"插入ID: {insert_result.inserted_id}")
批量插入
products = [
{"name": "Mechanical Keyboard", "price": 89.99},
{"name": "4K Monitor", "price": 349.99}
]
collection.insert_many(products)
读取数据（Read）
基础查询
for doc in collection.find({"price": {"$gt": 30}}):
print(doc["name"], "->", doc["price"])
投影操作（字段过滤）
cursor = collection.find(
{"tags": "electronics"},
{"name": 1, "price": 1, "_id": 0}
)
分页查询
page_size = 10
skip = (2 - 1) * page_size # 获取第二页
results = collection.find().skip(skip).limit(page_size)
更新数据（Update）
更新单个字段
collection.update_one(
{"name": "Wireless Mouse"},
{"$set": {"price": 34.99}}
)
数组操作（追加标签）
collection.update_one(
{"name": "4K Monitor"},
{"$addToSet": {"tags": {"$each": ["gaming", "professional"]}}}
)
批量更新（价格调整）
collection.update_many(
{"specs.dpi": {"$gt": 5000}},
{"$inc": {"price": 10}}
)
删除数据（Delete）
删除单个文档
collection.delete_one({"name": "Mechanical Keyboard"})
条件删除（清理测试数据）
collection.delete_many({"tags": {"$exists": False}})
清空集合
collection.drop()
高级查询技巧：解锁MongoDB的隐藏技能
复合查询示例
价格区间且包含特定规格
query = {
"price": {"$gte": 30, "$lte": 100},
"$or": [
{"specs.battery": "AAA x2"},
{"specs.battery": "Rechargeable"}
]
}
for doc in collection.find(query):
print(doc)
正则表达式匹配
模糊查询产品名称
pattern = re.compile(r'^Wire', re.IGNORECASE)
collection.find({"name": pattern})
聚合管道实战
pipeline = [
{"$match": {"price": {"$gt": 50}}},
{"$group": {
"_id": "$category",
"total_revenue": {"$sum": {"$multiply": ["$price", "$stock"]}}
}},
{"$sort": {"total_revenue": -1}},
{"$limit": 3}
]
result = collection.aggregate(pipeline)
索引优化：让查询飞起来
索引创建策略
单字段索引
collection.create_index("name")
复合索引
collection.create_index([("price", pymongo.ASCENDING), ("category", pymongo.DESCENDING)])
TTL索引（自动过期）
collection.create_index("created_at", expireAfterSeconds=3600)
索引性能分析
解释执行计划
explain_result = collection.find({"price": {"$gt": 30}}).explain()
print(explain_result["executionStats"]["executionTimeMillis"])
事务处理：保证数据一致性
会话管理示例
with client.start_session() as session:
try:
with session.start_transaction():
# 执行多个操作
collection.update_one(
{"_id": "order_123"},
{"$inc": {"total": 100}},
session=session
)
inventory.update_one(
{"product": "item_456"},
{"$inc": {"stock": -1}},
session=session
)
session.commit_transaction()
except Exception:
session.abort_transaction()
实际应用场景：电商系统实战
订单处理模块
def create_order(user_id, items):
with client.start_session() as session:
try:
with session.start_transaction():
# 创建订单
order_doc = {
"user_id": user_id,
"items": items,
"status": "pending",
"created_at": datetime.now()
}
order_id = orders.insert_one(order_doc, session=session).inserted_id
# 更新库存
for item in items:
inventory.update_one(
{"product_id": item["product_id"]},
{"$inc": {"stock": -item["quantity"]}},
session=session
)
session.commit_transaction()
return order_id
except Exception as e:
session.abort_transaction()
raise e
推荐系统集成
def get_recommendations(user_id):
pipeline = [
{"$match": {"user_id": user_id}},
{"$lookup": {
"from": "products",
"localField": "viewed_items",
"foreignField": "_id",
"as": "viewed_products"
}},
{"$unwind": "$viewed_products"},
{"$group": {
"_id": "$viewed_products.category",
"count": {"$sum": 1}
}},
{"$sort": {"count": -1}},
{"$limit": 3}
]
return list(analytics.aggregate(pipeline))
性能调优：从代码到架构的优化策略
批量写入优化
使用bulk_write提升性能
operations = [
UpdateOne({"sku": "A100"}, {"$inc": {"stock": -1}}),
UpdateOne({"sku": "B200"}, {"$inc": {"stock": -2}}),
InsertOne({"sku": "C300", "stock": 100})
]
result = collection.bulk_write(operations)
print(f"修改数量: {result.modified_count}")
连接池配置建议
生产环境推荐配置
client = MongoClient(
'mongodb://primary:27017,secondary:27017',
maxPoolSize=100,
minPoolSize=10,
waitQueueTimeoutMS=5000,
connectTimeoutMS=3000,
socketTimeoutMS=None
)
查询优化检查清单
确保查询字段都有对应索引
避免全集合扫描（$where操作符慎用）
使用projection减少数据传输量
合理设置batch_size控制内存使用
定期执行compact和repairDatabase维护
故障排查：常见问题解决方案
连接问题排查
try:
client.admin.command('ping')
except ConnectionFailure:
print("无法连接到MongoDB服务")
except OperationFailure as e:
print(f"认证失败: {str(e)}")
慢查询日志分析
启用分析器
db.setProfilingLevel(2, 100) # 记录所有超过100ms的操作
查询分析结果
for doc in db.system.profile.find().sort("ts": -1).limit(10):
print(f"{doc.millis}ms - {doc.op} - {doc.ns}")
死锁检测与处理
from pymongo.errors import PyMongoError
try:
# 执行可能冲突的操作
except PyMongoError as e:
if "transaction has been aborted" in str(e):
# 重试逻辑或记录警告
pass
总结：构建高效数据应用的七个原则
合理设计文档结构，避免过度嵌套
索引不是越多越好，定期审计索引使用
批量操作优先于循环单条操作
事务只用于必要场景，避免长事务
充分利用聚合管道代替应用层计算
连接池参数需根据负载动态调整
定期进行性能基线测试和优化
通过本文的实战代码和最佳实践，你已经掌握了Python操作MongoDB的核心技能。从简单的CRUD到复杂的事务处理，从索引优化到性能调优，这些知识将帮助你构建出高效可靠的数据驱动型应用。记住，最好的学习方式就是立即动手实践——现在就打开你的编辑器，开始第一个MongoDB项目吧！
