目录
Python快速上手（三十一）
Python MongoDB 详解
MongoDB 是一种 NoSQL 数据库，它使用文档存储数据，提供高性能、高可用性和易扩展性。Python 提供了 pymongo 库来与 MongoDB 进行交互。本文将详细讲解如何使用 Python 与 MongoDB 进行各种操作，包括连接数据库、CRUD 操作、查询、索引和聚合。
1. 安装 pymongo
在使用 MongoDB 前，需要安装 pymongo 库。可以使用以下命令安装：
pip install pymongo
2. 连接 MongoDB
2.1 基本连接
使用 MongoClient 类连接到 MongoDB 服务器：
from pymongo import MongoClient
client = MongoClient('localhost', 27017)
2.2 使用 URI 连接
可以使用 MongoDB URI 连接字符串：
client = MongoClient('mongodb://localhost:27017/')
2.3 访问数据库
连接到特定的数据库：
db = client['mydatabase']
或者：
db = client.mydatabase
2.4 认证
如果 MongoDB 需要认证，使用以下方式连接：
client = MongoClient('mongodb://username:password@localhost:27017/mydatabase')
db = client['mydatabase']
3. 创建和删除集合
3.1 创建集合
MongoDB 会在第一次插入文档时自动创建集合：
collection = db['mycollection']
3.2 删除集合
使用 drop 方法删除集合：
collection.drop()
4. CRUD 操作
4.1 插入文档
4.1.1 插入单个文档
使用 insert_one 方法插入一个文档：
document = {"name": "John", "age": 25, "city": "New York"}
collection.insert_one(document)
4.1.2 插入多个文档
使用 insert_many 方法插入多个文档：
documents = [
{"name": "Anna", "age": 28, "city": "London"},
{"name": "Mike", "age": 32, "city": "Chicago"}
]
collection.insert_many(documents)
4.2 查询文档
4.2.1 查询单个文档
使用 find_one 方法查询单个文档：
result = collection.find_one({"name": "John"})
print(result)
4.2.2 查询多个文档
使用 find 方法查询多个文档：
results = collection.find({"age": {"$gt": 20}})
for result in results:
print(result)
4.3 更新文档
4.3.1 更新单个文档
使用 update_one 方法更新单个文档：
collection.update_one({"name": "John"}, {"$set": {"age": 26}})
4.3.2 更新多个文档
使用 update_many 方法更新多个文档：
collection.update_many({"age": {"$gt": 20}}, {"$set": {"city": "San Francisco"}})
4.4 删除文档
4.4.1 删除单个文档
使用 delete_one 方法删除单个文档：
collection.delete_one({"name": "John"})
4.4.2 删除多个文档
使用 delete_many 方法删除多个文档：
collection.delete_many({"age": {"$gt": 20}})
5. 查询操作
5.1 条件查询
使用查询操作符进行条件查询：
results = collection.find({"age": {"$gte": 25}})
常用查询操作符：
- $eq：等于
- $ne：不等于
- $gt：大于
- $gte：大于等于
- $lt：小于
- $lte：小于等于
- $in：在指定数组内
- $nin：不在指定数组内
5.2 逻辑操作符
使用逻辑操作符进行查询：
results = collection.find({"$or": [{"age": {"$lt": 25}}, {"city": "London"}]})
常用逻辑操作符：
- $and：与
- $or：或
- $not：非
- $nor：非或
5.3 正则表达式
使用正则表达式进行查询：
results = collection.find({"name": {"$regex": "^J"}})
5.4 字段选择
指定返回的字段：
results = collection.find({}, {"_id": 0, "name": 1, "age": 1})
5.5 排序
使用 sort 方法进行排序：
results = collection.find().sort("age", -1)
5.6 限制和跳过
使用 limit 和 skip 方法进行分页：
results = collection.find().skip(5).limit(10)
6. 索引
6.1 创建索引
使用 create_index 方法创建索引：
collection.create_index([("name", 1)])
6.2 列出索引
使用 list_indexes 方法列出索引：
for index in collection.list_indexes():
print(index)
6.3 删除索引
使用 drop_index 方法删除索引：
collection.drop_index("name_1")
6.4 删除所有索引
使用 drop_indexes 方法删除所有索引：
collection.drop_indexes()
7. 聚合
7.1 基本聚合
使用 aggregate 方法进行聚合：
pipeline = [
{"$match": {"age": {"$gte": 25}}},
{"$group": {"_id": "$city", "average_age": {"$avg": "$age"}}}
]
results = collection.aggregate(pipeline)
for result in results:
print(result)
7.2 聚合操作符
常用聚合操作符：
- $match：过滤数据
- $group：分组并进行计算
- $sort：排序
- $limit：限制结果数量
- $skip：跳过指定数量的结果
- $project：改变输出文档的结构
- $unwind：拆分数组字段中的元素
7.3 聚合示例
计算每个城市的平均年龄：
pipeline = [
{"$group": {"_id": "$city", "average_age": {"$avg": "$age"}}}
]
results = collection.aggregate(pipeline)
for result in results:
print(result)
8. 其他操作
8.1 统计集合文档数量
使用 count_documents 方法统计文档数量：
count = collection.count_documents({"age": {"$gte": 25}})
print(count)
8.2 执行命令
使用 command 方法执行数据库命令：
result = db.command("serverStatus")
print(result)
9. 连接池和超时
9.1 设置连接池
可以设置连接池参数，例如最大连接数：
client = MongoClient('mongodb://localhost:27017/', maxPoolSize=50)
9.2 设置超时
可以设置连接超时和操作超时：
client = MongoClient('mongodb://localhost:27017/', serverSelectionTimeoutMS=5000, socketTimeoutMS=2000)
10. 实际应用案例
10.1 用户注册系统
以下示例展示了如何使用 MongoDB 实现一个简单的用户注册系统：
from pymongo import MongoClient
import datetime
client = MongoClient('mongodb://localhost:27017/')
db = client['user_database']
users = db['users']
# 注册用户
def register_user(username, password, email):
user = {
"username": username,
"password": password,
"email": email,
"created_at": datetime.datetime.now()
}
users.insert_one(user)
print(f"User {username} registered successfully")
# 查询用户
def find_user(username):
user = users.find_one({"username": username})
if user:
print(f"User found: {user}")
else:
print("User not found")
# 更新用户密码
def update_password(username, new_password):
result = users.update_one({"username": username}, {"$set": {"password": new_password}})
if result.matched_count:
print(f"Password updated for user {username}")
else:
print("User not found")
# 删除用户
def delete_user(username):
result = users.delete_one({"username": username})