4.8 Python 与 MongoDB 数据交互（pymongo）

达文汐

1179人浏览 · 2025-09-24 08:00:00

达文汐 · 2025-09-24 08:00:00 发布

引言：文档数据库的革命

在当今大数据和敏捷开发的时代，传统关系型数据库虽然强大，但在处理非结构化数据、快速迭代和水平扩展方面面临挑战。这就是 NoSQL 数据库诞生的背景，而 MongoDB 作为最流行的文档型数据库，以其灵活的文档模型、强大的查询能力和易扩展性，成为了现代应用开发的重要选择。

MongoDB 使用类似 JSON 的 BSON 格式存储数据，这种文档模型与 Python 的字典结构天然契合，使得 Python 成为与 MongoDB 交互的理想语言。本章将深入探讨如何使用 pymongo 库进行 MongoDB 数据库操作，从基础概念到高级特性，并通过实战项目展示如何构建基于 MongoDB 的现代应用。

第一部分：MongoDB 基础与 pymongo 入门

1.1 MongoDB 核心概念

与关系型数据库不同，MongoDB 使用以下概念：

MongoDB	关系型数据库	说明
数据库（Database）	数据库（Database）	数据容器
集合（Collection）	表（Table）	文档组
文档（Document）	行（Row）	数据记录
字段（Field）	列（Column）	数据属性
嵌入式文档	连接（Join）	嵌套数据结构

1.2 安装与配置

安装 MongoDB：

Windows：从 MongoDB 官网下载安装程序
macOS：使用 Homebrew：brew install mongodb-community

Linux (Ubuntu)：

wget -qO - https://www.mongodb.org/static/pgp/server-6.0.asc | sudo apt-key add -
echo "deb [ arch=amd64,arm64 ] https://repo.mongodb.org/apt/ubuntu focal/mongodb-org/6.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-6.0.list
sudo apt-get update
sudo apt-get install -y mongodb-org

安装 pymongo：

pip install pymongo

1.3 连接 MongoDB

import pymongo
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure, OperationFailure

def create_connection():
    """创建MongoDB连接"""
    try:
        # 连接到本地MongoDB实例
        client = MongoClient('mongodb://localhost:27017/')
        
        # 测试连接
        client.admin.command('ping')
        print("成功连接到MongoDB")
        
        return client
    except ConnectionFailure as e:
        print(f"连接MongoDB失败: {e}")
        return None
    except Exception as e:
        print(f"发生未知错误: {e}")
        return None

# 高级连接选项
def create_advanced_connection():
    """创建带配置的MongoDB连接"""
    try:
        client = MongoClient(
            host='localhost',
            port=27017,
            username='your_username',  # 如果启用了认证
            password='your_password',
            authSource='admin',        # 认证数据库
            connectTimeoutMS=30000,    # 连接超时30秒
            socketTimeoutMS=30000,     # 套接字超时30秒
            serverSelectionTimeoutMS=30000,  # 服务器选择超时30秒
            maxPoolSize=50,            # 最大连接池大小
            minPoolSize=10,            # 最小连接池大小
            retryWrites=True           # 启用重试写入
        )
        
        # 测试连接
        client.admin.command('ping')
        print("高级连接创建成功")
        
        return client
    except Exception as e:
        print(f"创建连接时发生错误: {e}")
        return None

# 连接到副本集或分片集群
def connect_to_replica_set():
    """连接到MongoDB副本集"""
    try:
        client = MongoClient(
            'mongodb://host1:27017,host2:27017,host3:27017/',
            replicaSet='myReplicaSet',
            readPreference='secondaryPreferred'  # 优先从 secondary 节点读取
        )
        
        client.admin.command('ping')
        print("成功连接到副本集")
        
        return client
    except Exception as e:
        print(f"连接副本集失败: {e}")
        return None

# 测试连接
client = create_connection()
if client:
    # 获取数据库列表
    print("数据库列表:", client.list_database_names())
    client.close()

第二部分：数据库与集合操作

2.1 数据库操作

def database_operations():
    """数据库操作示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        # 获取或创建数据库
        db = client['company']  # 如果不存在会自动创建
        print(f"使用数据库: {db.name}")
        
        # 获取所有集合名称
        collections = db.list_collection_names()
        print("集合列表:", collections)
        
        # 检查集合是否存在
        if 'employees' in collections:
            print("employees集合已存在")
        else:
            print("employees集合不存在")
        
        # 获取数据库统计信息
        stats = db.command('dbStats')
        print(f"数据库大小: {stats['dataSize']} 字节")
        print(f"文档数量: {stats['objects']}")
        
    except Exception as e:
        print(f"数据库操作时发生错误: {e}")
    finally:
        client.close()

# 执行数据库操作
database_operations()

2.2 集合操作

def collection_operations():
    """集合操作示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        db = client['company']
        
        # 获取或创建集合
        employees_collection = db['employees']
        print(f"使用集合: {employees_collection.name}")
        
        # 创建带选项的集合
        try:
            # 创建带验证规则的集合
            db.create_collection(
                'departments',
                validator={
                    '$jsonSchema': {
                        'bsonType': 'object',
                        'required': ['name', 'location'],
                        'properties': {
                            'name': {
                                'bsonType': 'string',
                                'description': '必须是一个字符串'
                            },
                            'location': {
                                'bsonType': 'string',
                                'description': '必须是一个字符串'
                            },
                            'budget': {
                                'bsonType': 'double',
                                'description': '必须是一个数字'
                            }
                        }
                    }
                }
            )
            print("带验证规则的departments集合创建成功")
        except Exception as e:
            print(f"创建集合时发生错误: {e}")
        
        # 获取集合统计信息
        stats = db.command('collStats', 'employees')
        print(f"集合大小: {stats['size']} 字节")
        print(f"文档数量: {stats['count']}")
        
        # 重命名集合
        # db['old_collection'].rename('new_collection')
        
        # 删除集合
        # db['temp_collection'].drop()
        
    except Exception as e:
        print(f"集合操作时发生错误: {e}")
    finally:
        client.close()

# 执行集合操作
collection_operations()

第三部分：文档 CRUD 操作

3.1 插入文档（Create）

def insert_documents():
    """插入文档示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        db = client['company']
        employees = db['employees']
        departments = db['departments']
        
        # 插入单个文档
        department_data = {
            'name': '技术部',
            'location': '北京',
            'budget': 1000000.00,
            'created_at': datetime.now()
        }
        
        dept_result = departments.insert_one(department_data)
        print(f"插入部门成功，ID: {dept_result.inserted_id}")
        
        # 插入多个文档
        employees_data = [
            {
                'first_name': '张',
                'last_name': '三',
                'email': 'zhangsan@email.com',
                'phone': '13800138001',
                'hire_date': datetime(2020, 1, 15),
                'salary': 15000.00,
                'department': dept_result.inserted_id,
                'skills': ['Python', 'JavaScript', 'SQL'],
                'address': {
                    'street': '海淀区中关村大街',
                    'city': '北京',
                    'zipcode': '100000'
                },
                'created_at': datetime.now()
            },
            {
                'first_name': '李',
                'last_name': '四',
                'email': 'lisi@email.com',
                'phone': '13800138002',
                'hire_date': datetime(2019, 3, 10),
                'salary': 12000.00,
                'department': dept_result.inserted_id,
                'skills': ['Java', 'Spring', 'MySQL'],
                'address': {
                    'street': '朝阳区国贸',
                    'city': '北京',
                    'zipcode': '100000'
                },
                'created_at': datetime.now()
            }
        ]
        
        emp_result = employees.insert_many(employees_data)
        print(f"插入员工成功，ID列表: {emp_result.inserted_ids}")
        
        # 插入带自定义ID的文档
        custom_id_doc = {
            '_id': 'emp_001',
            'first_name': '王',
            'last_name': '五',
            'email': 'wangwu@email.com',
            'created_at': datetime.now()
        }
        
        custom_result = employees.insert_one(custom_id_doc)
        print(f"插入自定义ID文档成功，ID: {custom_result.inserted_id}")
        
    except Exception as e:
        print(f"插入文档时发生错误: {e}")
    finally:
        client.close()

# 执行插入操作
insert_documents()

3.2 查询文档（Read）

def query_documents():
    """查询文档示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        db = client['company']
        employees = db['employees']
        
        # 查询所有文档
        print("所有员工:")
        all_employees = employees.find()
        for emp in all_employees:
            print(f"{emp['first_name']} {emp['last_name']} - {emp['email']}")
        
        # 条件查询
        print("\n薪资大于10000的员工:")
        well_paid_employees = employees.find({'salary': {'$gt': 10000}})
        for emp in well_paid_employees:
            print(f"{emp['first_name']} {emp['last_name']}: ￥{emp['salary']:,.2f}")
        
        # 多条件查询
        print("\n在北京的技术部员工:")
        tech_employees = employees.find({
            'salary': {'$gt': 10000},
            'address.city': '北京'
        })
        for emp in tech_employees:
            print(f"{emp['first_name']} {emp['last_name']}")
        
        # 投影（选择返回的字段）
        print("\n只返回姓名和邮箱:")
        name_email_only = employees.find(
            {},  # 空查询条件表示所有文档
            {'first_name': 1, 'last_name': 1, 'email': 1, '_id': 0}  # 1表示包含，0表示排除
        )
        for emp in name_email_only:
            print(emp)
        
        # 排序和限制
        print("\n薪资最高的3名员工:")
        top_earners = employees.find().sort('salary', -1).limit(3)
        for emp in top_earners:
            print(f"{emp['first_name']} {emp['last_name']}: ￥{emp['salary']:,.2f}")
        
        # 统计文档数量
        count = employees.count_documents({'salary': {'$gt': 10000}})
        print(f"\n薪资大于10000的员工数量: {count}")
        
        # 查询单个文档
        zhangsan = employees.find_one({'first_name': '张', 'last_name': '三'})
        if zhangsan:
            print(f"\n找到张三: {zhangsan['email']}")
        
    except Exception as e:
        print(f"查询文档时发生错误: {e}")
    finally:
        client.close()

# 执行查询操作
query_documents()

3.3 更新文档（Update）

def update_documents():
    """更新文档示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        db = client['company']
        employees = db['employees']
        
        # 更新单个文档
        update_result = employees.update_one(
            {'first_name': '张', 'last_name': '三'},
            {'$set': {'salary': 16000.00, 'last_updated': datetime.now()}}
        )
        print(f"匹配文档数: {update_result.matched_count}, 修改文档数: {update_result.modified_count}")
        
        # 更新多个文档
        update_many_result = employees.update_many(
            {'salary': {'$lt': 15000}},
            {'$inc': {'salary': 1000}, '$set': {'last_updated': datetime.now()}}
        )
        print(f"薪资调整影响员工数: {update_many_result.modified_count}")
        
        # 使用更新操作符
        # $set: 设置字段值
        # $unset: 删除字段
        # $inc: 增加字段值
        # $push: 向数组添加元素
        # $pull: 从数组移除元素
        
        # 添加技能到员工
        employees.update_one(
            {'first_name': '张', 'last_name': '三'},
            {'$push': {'skills': 'MongoDB'}}
        )
        print("已添加MongoDB技能")
        
        # 替换整个文档
        replacement = {
            'first_name': '张',
            'last_name': '三',
            'email': 'zhangsan_new@email.com',
            'position': '高级工程师',
            'updated_at': datetime.now()
        }
        
        replace_result = employees.replace_one(
            {'first_name': '张', 'last_name': '三'},
            replacement
        )
        print(f"文档替换结果: 匹配{replace_result.matched_count}个, 修改{replace_result.modified_count}个")
        
    except Exception as e:
        print(f"更新文档时发生错误: {e}")
    finally:
        client.close()

# 执行更新操作
update_documents()

3.4 删除文档（Delete）

def delete_documents():
    """删除文档示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        db = client['company']
        employees = db['employees']
        
        # 先插入一些测试数据
        test_employee = {
            'first_name': 'Test',
            'last_name': 'User',
            'email': 'test@example.com',
            'created_at': datetime.now()
        }
        test_result = employees.insert_one(test_employee)
        print(f"插入测试员工成功，ID: {test_result.inserted_id}")
        
        # 删除单个文档
        delete_result = employees.delete_one({'email': 'test@example.com'})
        print(f"删除测试员工，删除文档数: {delete_result.deleted_count}")
        
        # 删除多个文档（谨慎使用！）
        # 先标记一些文档为待删除
        employees.update_many(
            {'salary': {'$lt': 10000}},
            {'$set': {'to_delete': True}}
        )
        
        # 删除标记的文档
        delete_many_result = employees.delete_many({'to_delete': True})
        print(f"删除低薪资员工，删除文档数: {delete_many_result.deleted_count}")
        
        # 更安全的做法：标记删除而不是物理删除
        employees.update_many(
            {'salary': {'$lt': 10000}},
            {'$set': {'active': False, 'inactive_date': datetime.now()}}
        )
        print("已标记低薪资员工为不活跃状态")
        
        # 删除所有文档（非常危险！）
        # delete_all_result = employees.delete_many({})
        # print(f"删除所有文档，删除文档数: {delete_all_result.deleted_count}")
        
    except Exception as e:
        print(f"删除文档时发生错误: {e}")
    finally:
        client.close()

# 执行删除操作
delete_documents()

第四部分：高级查询与聚合操作

4.1 复杂查询操作

def complex_queries():
    """复杂查询示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        db = client['company']
        employees = db['employees']
        
        # 数组查询
        print("会Python的员工:")
        python_devs = employees.find({'skills': 'Python'})
        for emp in python_devs:
            print(f"{emp['first_name']} {emp['last_name']}: {emp['skills']}")
        
        # 多技能查询
        print("\n既会Python又会JavaScript的员工:")
        full_stack_devs = employees.find({'skills': {'$all': ['Python', 'JavaScript']}})
        for emp in full_stack_devs:
            print(f"{emp['first_name']} {emp['last_name']}")
        
        # 正则表达式查询
        print("\n邮箱是Gmail的员工:")
        gmail_users = employees.find({'email': {'$regex': '@gmail\\.com$'}})
        for emp in gmail_users:
            print(f"{emp['first_name']} {emp['last_name']}: {emp['email']}")
        
        # 嵌套文档查询
        print("\n在北京的员工:")
        beijing_employees = employees.find({'address.city': '北京'})
        for emp in beijing_employees:
            print(f"{emp['first_name']} {emp['last_name']}: {emp['address']['city']}")
        
        # 范围查询
        print("\n在2020年入职的员工:")
        start_2020 = datetime(2020, 1, 1)
        end_2020 = datetime(2020, 12, 31)
        employees_2020 = employees.find({
            'hire_date': {
                '$gte': start_2020,
                '$lte': end_2020
            }
        })
        for emp in employees_2020:
            print(f"{emp['first_name']} {emp['last_name']}: {emp['hire_date']}")
        
        # 或查询
        print("\n薪资大于15000或在技术部的员工:")
        high_salary_or_tech = employees.find({
            '$or': [
                {'salary': {'$gt': 15000}},
                {'department': {'$exists': True}}  # 假设技术部文档有department字段
            ]
        })
        for emp in high_salary_or_tech:
            print(f"{emp['first_name']} {emp['last_name']}: ￥{emp.get('salary', 0):,.2f}")
        
    except Exception as e:
        print(f"复杂查询时发生错误: {e}")
    finally:
        client.close()

# 执行复杂查询
complex_queries()

4.2 聚合框架

MongoDB 的聚合框架提供了强大的数据处理能力：

def aggregation_examples():
    """聚合操作示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        db = client['company']
        employees = db['employees']
        
        # 简单的分组聚合
        print("按城市统计员工数量:")
        pipeline_city = [
            {'$group': {
                '_id': '$address.city',
                'count': {'$sum': 1},
                'avg_salary': {'$avg': '$salary'},
                'max_salary': {'$max': '$salary'},
                'min_salary': {'$min': '$salary'}
            }},
            {'$sort': {'count': -1}}
        ]
        
        city_stats = employees.aggregate(pipeline_city)
        for stat in city_stats:
            print(f"{stat['_id']}: {stat['count']}人, 平均工资: ￥{stat['avg_salary']:,.2f}")
        
        # 多阶段聚合
        print("\n技能统计:")
        pipeline_skills = [
            {'$unwind': '$skills'},  # 将数组拆分为多个文档
            {'$group': {
                '_id': '$skills',
                'count': {'$sum': 1},
                'practitioners': {'$push': '$first_name'}
            }},
            {'$sort': {'count': -1}},
            {'$limit': 5}
        ]
        
        skills_stats = employees.aggregate(pipeline_skills)
        for stat in skills_stats:
            print(f"{stat['_id']}: {stat['count']}人掌握")
        
        # 复杂聚合：按部门统计
        print("\n部门统计:")
        pipeline_department = [
            {'$lookup': {  # 类似SQL的JOIN
                'from': 'departments',
                'localField': 'department',
                'foreignField': '_id',
                'as': 'dept_info'
            }},
            {'$unwind': '$dept_info'},
            {'$group': {
                '_id': '$dept_info.name',
                'employee_count': {'$sum': 1},
                'total_salary': {'$sum': '$salary'},
                'avg_salary': {'$avg': '$salary'},
                'employees': {'$push': {
                    'name': {'$concat': ['$first_name', ' ', '$last_name']},
                    'salary': '$salary'
                }}
            }},
            {'$sort': {'total_salary': -1}}
        ]
        
        dept_stats = employees.aggregate(pipeline_department)
        for stat in dept_stats:
            print(f"{stat['_id']}: {stat['employee_count']}人, 总工资: ￥{stat['total_salary']:,.2f}")
        
        # 使用聚合管道进行数据转换
        print("\n薪资等级分布:")
        pipeline_salary_brackets = [
            {'$bucket': {
                'groupBy': '$salary',
                'boundaries': [0, 10000, 15000, 20000, 30000],
                'default': '30000+',
                'output': {
                    'count': {'$sum': 1},
                    'employees': {'$push': '$first_name'}
                }
            }},
            {'$sort': {'_id': 1}}
        ]
        
        salary_brackets = employees.aggregate(pipeline_salary_brackets)
        for bracket in salary_brackets:
            print(f"薪资范围 {bracket['_id']}: {bracket['count']}人")
        
    except Exception as e:
        print(f"聚合操作时发生错误: {e}")
    finally:
        client.close()

# 执行聚合操作
aggregation_examples()

4.3 索引优化

def index_operations():
    """索引操作示例"""
    client = create_connection()
    if not client:
        return
    
    try:
        db = client['company']
        employees = db['employees']
        
        # 创建单字段索引
        employees.create_index([('email', pymongo.ASCENDING)], unique=True)
        print("已创建email唯一索引")
        
        # 创建复合索引
        employees.create_index([
            ('department', pymongo.ASCENDING),
            ('salary', pymongo.DESCENDING)
        ])
        print("已创建部门-薪资复合索引")
        
        # 创建文本索引（用于全文搜索）
        employees.create_index([
            ('first_name', pymongo.TEXT),
            ('last_name', pymongo.TEXT),
            ('skills', pymongo.TEXT)
        ])
        print("已创建文本索引")
        
        # 查看索引信息
        indexes = employees.list_indexes()
        print("\n集合索引:")
        for index in indexes:
            print(f"索引名称: {index['name']}, 键: {index['key']}")
        
        # 使用文本搜索
        print("\n搜索'Python'相关的员工:")
        text_results = employees.find({
            '$text': {'$search': 'Python'}
        })
        for emp in text_results:
            print(f"{emp['first_name']} {emp['last_name']}")
        
        # 删除索引
        # employees.drop_index('email_1')
        # print("已删除email索引")
        
    except Exception as e:
        print(f"索引操作时发生错误: {e}")
    finally:
        client.close()

# 执行索引操作
index_operations()

第五部分：综合实战项目——博客平台后端

现在让我们构建一个完整的博客平台后端，使用 MongoDB 存储数据。

项目功能：

用户管理
文章发布与管理
评论系统
标签分类
搜索功能

代码实现：

# blog_platform.py
import pymongo
from datetime import datetime, timedelta
from bson import ObjectId
from pymongo import MongoClient, DESCENDING, ASCENDING
from pymongo.errors import DuplicateKeyError, OperationFailure
import re

class BlogPlatform:
    def __init__(self, db_name='blog_db'):
        self.client = MongoClient('mongodb://localhost:27017/')
        self.db = self.client[db_name]
        self.init_database()
    
    def init_database(self):
        """初始化数据库结构和索引"""
        # 创建集合（如果不存在会自动创建）
        self.users = self.db['users']
        self.posts = self.db['posts']
        self.comments = self.db['comments']
        self.categories = self.db['categories']
        
        # 创建索引
        try:
            # 用户集合索引
            self.users.create_index('email', unique=True)
            self.users.create_index('username', unique=True)
            
            # 文章集合索引
            self.posts.create_index('author_id')
            self.posts.create_index('category_id')
            self.posts.create_index('tags')
            self.posts.create_index('created_at')
            self.posts.create_index([
                ('title', 'text'),
                ('content', 'text'),
                ('tags', 'text')
            ], name='search_index')
            
            # 评论集合索引
            self.comments.create_index('post_id')
            self.comments.create_index('author_id')
            self.comments.create_index('created_at')
            
            # 分类集合索引
            self.categories.create_index('name', unique=True)
            
            print("数据库初始化完成")
        except Exception as e:
            print(f"初始化数据库时发生错误: {e}")
    
    def create_user(self, username, email, password_hash, display_name=None):
        """创建用户"""
        try:
            user_data = {
                'username': username,
                'email': email,
                'password_hash': password_hash,
                'display_name': display_name or username,
                'created_at': datetime.now(),
                'last_login': None,
                'is_active': True,
                'role': 'user',  # user, author, admin
                'profile': {
                    'bio': '',
                    'avatar_url': None,
                    'website': None
                }
            }
            
            result = self.users.insert_one(user_data)
            print(f"用户 {username} 创建成功，ID: {result.inserted_id}")
            return result.inserted_id
        except DuplicateKeyError:
            print(f"用户名或邮箱已存在: {username}/{email}")
            return None
        except Exception as e:
            print(f"创建用户时发生错误: {e}")
            return None
    
    def create_post(self, author_id, title, content, category_id=None, tags=None):
        """创建博客文章"""
        try:
            post_data = {
                'title': title,
                'content': content,
                'author_id': author_id,
                'category_id': category_id,
                'tags': tags or [],
                'created_at': datetime.now(),
                'updated_at': datetime.now(),
                'published': True,
                'published_at': datetime.now(),
                'views': 0,
                'likes': 0,
                'comments_count': 0,
                'slug': self.generate_slug(title),
                'meta': {
                    'description': content[:150] + '...' if len(content) > 150 else content,
                    'keywords': tags or []
                }
            }
            
            result = self.posts.insert_one(post_data)
            print(f"文章 '{title}' 创建成功，ID: {result.inserted_id}")
            return result.inserted_id
        except Exception as e:
            print(f"创建文章时发生错误: {e}")
            return None
    
    def generate_slug(self, title):
        """生成URL友好的slug"""
        slug = re.sub(r'[^\w\s-]', '', title.lower())
        slug = re.sub(r'[-\s]+', '-', slug).strip('-')
        return slug
    
    def add_comment(self, post_id, author_id, content, parent_id=None):
        """添加评论"""
        try:
            comment_data = {
                'post_id': post_id,
                'author_id': author_id,
                'content': content,
                'parent_id': parent_id,  # 用于回复评论
                'created_at': datetime.now(),
                'updated_at': datetime.now(),
                'likes': 0,
                'is_approved': True
            }
            
            result = self.comments.insert_one(comment_data)
            
            # 更新文章的评论计数
            self.posts.update_one(
                {'_id': post_id},
                {'$inc': {'comments_count': 1}}
            )
            
            print(f"评论添加成功，ID: {result.inserted_id}")
            return result.inserted_id
        except Exception as e:
            print(f"添加评论时发生错误: {e}")
            return None
    
    def get_recent_posts(self, limit=10, page=1):
        """获取最近的文章"""
        try:
            skip = (page - 1) * limit
            posts = self.posts.find(
                {'published': True},
                {'content': 0}  # 不返回内容，提高性能
            ).sort('created_at', DESCENDING).skip(skip).limit(limit)
            
            # 填充作者信息
            posts_list = []
            for post in posts:
                author = self.users.find_one(
                    {'_id': post['author_id']},
                    {'display_name': 1, 'username': 1}
                )
                post['author'] = author
                posts_list.append(post)
            
            return posts_list
        except Exception as e:
            print(f"获取文章时发生错误: {e}")
            return []
    
    def get_post_by_slug(self, slug):
        """根据slug获取文章详情"""
        try:
            post = self.posts.find_one({'slug': slug})
            if not post:
                return None
            
            # 增加浏览次数
            self.posts.update_one(
                {'_id': post['_id']},
                {'$inc': {'views': 1}}
            )
            
            # 获取作者信息
            author = self.users.find_one(
                {'_id': post['author_id']},
                {'display_name': 1, 'username': 1, 'profile': 1}
            )
            post['author'] = author
            
            # 获取评论
            comments = self.comments.find(
                {'post_id': post['_id'], 'is_approved': True}
            ).sort('created_at', ASCENDING)
            
            # 获取评论作者信息
            comments_list = []
            for comment in comments:
                comment_author = self.users.find_one(
                    {'_id': comment['author_id']},
                    {'display_name': 1, 'username': 1}
                )
                comment['author'] = comment_author
                comments_list.append(comment)
            
            post['comments'] = comments_list
            
            return post
        except Exception as e:
            print(f"获取文章详情时发生错误: {e}")
            return None
    
    def search_posts(self, query, limit=10, page=1):
        """搜索文章"""
        try:
            skip = (page - 1) * limit
            
            # 文本搜索
            results = self.posts.find(
                {
                    '$text': {'$search': query},
                    'published': True
                },
                {
                    'score': {'$meta': 'textScore'},
                    'content': 0
                }
            ).sort([('score', {'$meta': 'textScore'})]).skip(skip).limit(limit)
            
            posts_list = []
            for post in results:
                author = self.users.find_one(
                    {'_id': post['author_id']},
                    {'display_name': 1, 'username': 1}
                )
                post['author'] = author
                posts_list.append(post)
            
            return posts_list
        except Exception as e:
            print(f"搜索文章时发生错误: {e}")
            return []
    
    def get_popular_posts(self, limit=5):
        """获取热门文章"""
        try:
            pipeline = [
                {'$match': {'published': True}},
                {'$sort': {'views': DESCENDING}},
                {'$limit': limit},
                {'$project': {
                    'title': 1,
                    'slug': 1,
                    'views': 1,
                    'created_at': 1,
                    'author_id': 1
                }}
            ]
            
            posts = self.posts.aggregate(pipeline)
            
            posts_list = []
            for post in posts:
                author = self.users.find_one(
                    {'_id': post['author_id']},
                    {'display_name': 1}
                )
                post['author'] = author
                posts_list.append(post)
            
            return posts_list
        except Exception as e:
            print(f"获取热门文章时发生错误: {e}")
            return []
    
    def get_posts_by_tag(self, tag, limit=10, page=1):
        """根据标签获取文章"""
        try:
            skip = (page - 1) * limit
            posts = self.posts.find(
                {
                    'tags': tag,
                    'published': True
                },
                {'content': 0}
            ).sort('created_at', DESCENDING).skip(skip).limit(limit)
            
            posts_list = []
            for post in posts:
                author = self.users.find_one(
                    {'_id': post['author_id']},
                    {'display_name': 1, 'username': 1}
                )
                post['author'] = author
                posts_list.append(post)
            
            return posts_list
        except Exception as e:
            print(f"根据标签获取文章时发生错误: {e}")
            return []
    
    def get_category_stats(self):
        """获取分类统计"""
        try:
            pipeline = [
                {'$match': {'published': True}},
                {'$group': {
                    '_id': '$category_id',
                    'post_count': {'$sum': 1},
                    'total_views': {'$sum': '$views'},
                    'total_comments': {'$sum': '$comments_count'}
                }},
                {'$lookup': {
                    'from': 'categories',
                    'localField': '_id',
                    'foreignField': '_id',
                    'as': 'category_info'
                }},
                {'$unwind': '$category_info'},
                {'$project': {
                    'category_name': '$category_info.name',
                    'post_count': 1,
                    'total_views': 1,
                    'total_comments': 1
                }},
                {'$sort': {'post_count': DESCENDING}}
            ]
            
            stats = self.posts.aggregate(pipeline)
            return list(stats)
        except Exception as e:
            print(f"获取分类统计时发生错误: {e}")
            return []
    
    def generate_sample_data(self):
        """生成示例数据"""
        try:
            # 创建示例用户
            admin_id = self.create_user('admin', 'admin@blog.com', 'hashed_password', '管理员')
            author1_id = self.create_user('author1', 'author1@blog.com', 'hashed_password', '作者一')
            author2_id = self.create_user('author2', 'author2@blog.com', 'hashed_password', '作者二')
            
            # 创建分类
            categories = [
                {'name': '技术', 'description': '技术相关文章'},
                {'name': '生活', 'description': '生活相关文章'},
                {'name': '旅行', 'description': '旅行相关文章'}
            ]
            
            category_ids = []
            for category in categories:
                result = self.categories.insert_one(category)
                category_ids.append(result.inserted_id)
            
            # 创建示例文章
            posts = [
                {
                    'author_id': author1_id,
                    'title': 'Python MongoDB 教程',
                    'content': '这是一篇关于如何使用Python操作MongoDB的详细教程...',
                    'category_id': category_ids[0],
                    'tags': ['Python', 'MongoDB', '数据库']
                },
                {
                    'author_id': author1_id,
                    'title': 'Web开发最佳实践',
                    'content': '本文介绍了现代Web开发的最佳实践和模式...',
                    'category_id': category_ids[0],
                    'tags': ['Web开发', '最佳实践', '编程']
                },
                {
                    'author_id': author2_id,
                    'title': '我的西藏之旅',
                    'content': '分享我在西藏旅行的经历和见闻...',
                    'category_id': category_ids[2],
                    'tags': ['旅行', '西藏', '摄影']
                }
            ]
            
            for post_data in posts:
                self.create_post(**post_data)
            
            print("示例数据生成成功")
            
        except Exception as e:
            print(f"生成示例数据时发生错误: {e}")

# 使用示例
def main():
    # 创建博客平台实例
    blog = BlogPlatform()
    
    # 生成示例数据
    blog.generate_sample_data()
    
    # 获取最近文章
    print("最近文章:")
    recent_posts = blog.get_recent_posts(limit=5)
    for post in recent_posts:
        print(f"- {post['title']} by {post['author']['display_name']}")
    
    # 搜索文章
    print("\n搜索'Python'相关文章:")
    search_results = blog.search_posts('Python')
    for post in search_results:
        print(f"- {post['title']} (相关度: {post.get('score', 0):.2f})")
    
    # 获取热门文章
    print("\n热门文章:")
    popular_posts = blog.get_popular_posts()
    for post in popular_posts:
        print(f"- {post['title']} ({post['views']} 次阅读)")
    
    # 获取分类统计
    print("\n分类统计:")
    category_stats = blog.get_category_stats()
    for stat in category_stats:
        print(f"{stat['category_name']}: {stat['post_count']}篇文章, {stat['total_views']}次阅读")

if __name__ == '__main__':
    main()

项目扩展思路：

用户认证系统：添加JWT token认证
文件上传：集成图片和文件上传功能
缓存系统：使用Redis缓存热门文章和查询结果
API接口：创建RESTful API供前端应用调用
后台管理：构建文章管理和用户管理后台
SEO优化：添加sitemap和meta标签优化
社交媒体集成：添加社交分享和评论功能

总结与最佳实践

通过本章的学习，你已经全面掌握了使用pymongo进行MongoDB数据库操作的各个方面：

MongoDB基础：了解了文档数据库的概念和特点
pymongo连接：掌握了连接MongoDB的各种方式和配置选项
CRUD操作：深入理解了文档的增删改查操作
高级查询：学会了复杂查询、聚合管道和索引优化
实战项目：构建了完整的博客平台后端，综合运用了所学知识

最佳实践总结：

合理设计文档结构，避免过度嵌套和数组过大
为常用查询字段创建合适的索引
使用投影优化查询性能，只返回需要的字段
合理使用聚合管道进行复杂数据处理
实施适当的错误处理和重试机制
定期监控和优化数据库性能
使用连接池管理数据库连接
实施适当的数据备份和恢复策略

MongoDB作为最流行的NoSQL数据库，与Python的结合为开发现代Web应用、移动应用和大数据处理系统提供了强大的数据存储和处理能力。通过掌握pymongo的使用，你能够构建灵活、可扩展和高性能的数据驱动应用。

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

SeedVR2终极指南：如何用单步扩散模型实现专业级视频修复

还在为老旧视频的模糊画质而烦恼吗？面对抖动严重的监控录像或褪色的家庭影像，你是否曾希望有一种简单高效的修复方案？SeedVR2正是为解决这些痛点而生的一站式视频修复工具。作为字节跳动推出的先进视频修复模型，它能够在单步推理中实现专业级的视频质量提升，彻底改变了传统视频修复的工作流程。## 核心价值：为什么选择SeedVR2进行视频修复？**传统视频修复的三大瓶颈**：1. **计算成本高

AI Agent技术社区

深度解析Amlogic S9xxx电视盒子Armbian系统：从闲置设备到Linux服务器的终极实战指南

你是否拥有一台闲置的Amlogic S9xxx系列电视盒子？想让它摆脱单一的视频播放功能，变身为功能强大的Linux服务器、开发平台或家庭媒体中心？本指南将为你详细解析如何通过Armbian系统实现这一转变，让你的电视盒子重获新生，成为高性能的Linux设备。Armbian系统基于Debian/Ubuntu构建，专为ARM芯片优化，提供了完整的Linux生态支持。## 🎯 核心价值与独特优势

AI Agent技术社区

CANN/runtime Event管理API

本章节描述 CANN Runtime 的 Event 管理接口，用于事件的创建、记录、同步、计时及 IPC 跨进程共享。- [`aclError aclrtCreateEvent(aclrtEvent *event)`](#aclrtCreateEvent)：创建Event，创建出来的Event可用于统计两个Event之间的耗时、多Stream之间的任务同步等场景。- [`aclError