Python 连接 MongoDB 完整指南：从连接配置到增删改查实战

前言

在现代应用开发中，MongoDB 作为一款流行的 NoSQL 数据库，以其灵活的文档模型、强大的查询能力和优秀的性能，成为了众多开发者的首选。本文将基于实际项目代码，详细介绍如何使用 Python 连接 MongoDB，并实现数据的插入、查询、更新和删除等核心操作。

环境准备

1. 安装依赖

首先需要安装 MongoDB 的 Python 驱动 pymongo：

pip install pymongo
pip install python-dotenv

2. 配置环境变量

为了安全起见，我们将数据库连接信息配置在 .env 文件中，避免硬编码敏感信息：

# MongoDB 连接配置
MONGO_URL=mongodb://root:123456@127.0.0.1:27017
MONGO_DB_NAME=test002

连接字符串格式说明：

mongodb://用户名:密码@主机:端口/数据库

完整代码实现

1. 导入必要模块

import os
import logging
from typing import List, Dict, Any, Optional
from datetime import datetime
from pymongo import MongoClient, ASCENDING
from bson import ObjectId
from dotenv import load_dotenv

# 加载 .env 文件中的环境变量
load_dotenv()

2. 创建 MongoDB 连接工具类

class HistoryMongoTool:
    """
    MongoDB 历史对话记录读写工具类
    核心功能：封装 MongoDB 的连接、集合初始化、索引创建
    """
    def __init__(self):
        try:
            # 从环境变量读取 MongoDB 连接地址
            self.mongo_url = os.getenv("MONGO_URL")
            # 从环境变量读取数据库名称
            self.db_name = os.getenv("MONGO_DB_NAME")
            
            # 创建 MongoDB 客户端实例，建立连接
            self.client = MongoClient(self.mongo_url)
            # 获取指定名称的数据库对象
            self.db = self.client[self.db_name]
            # 获取对话记录集合（相当于关系型数据库的表）
            self.chat_message = self.db["chat_message"]
            
            # 创建复合索引，提升查询性能
            # 索引规则：session_id 升序 + ts 降序
            self.chat_message.create_index([("session_id", 1), ("ts", -1)])
            
            logging.info(f"Successfully connected to MongoDB: {self.db_name}")
        except Exception as e:
            logging.error(f"Failed to connect to MongoDB: {e}")
            raise

3. 实现单例模式

为了避免重复创建数据库连接，我们使用单例模式：

# 全局变量：存储单例实例
_history_mongo_tool = None

# 模块加载时尝试初始化单例实例
try:
    _history_mongo_tool = HistoryMongoTool()
except Exception as e:
    logging.warning(f"Could not initialize HistoryMongoTool on module load: {e}")

def get_history_mongo_tool() -> HistoryMongoTool:
    """
    获取 HistoryMongoTool 的单例实例（懒加载模式）
    """
    global _history_mongo_tool
    if _history_mongo_tool is None:
        _history_mongo_tool = HistoryMongoTool()
    return _history_mongo_tool

核心操作详解

一、插入数据（Create）

1. 单条插入

def save_chat_message(
        session_id: str,
        role: str,
        text: str,
        rewritten_query: str = "",
        item_names: List[str] = None,
        image_urls: List[str] = None,
        message_id: str = None
) -> str:
    """
    写入/更新单条会话记录到 MongoDB
    :param session_id: 会话唯一标识
    :param role: 消息角色（user/assistant）
    :param text: 对话内容
    :param rewritten_query: 重写后的查询语句
    :param item_names: 关联的商品名称列表
    :param image_urls: 关联的图片 URL 列表
    :param message_id: 记录主键 ID（有则更新，无则新增）
    :return: 插入/更新的记录唯一标识
    """
    # 生成当前时间的时间戳
    ts = datetime.now().timestamp()
    
    # 构造文档数据
    document = {
        "session_id": session_id,
        "role": role,
        "text": text,
        "rewritten_query": rewritten_query or "",
        "item_names": item_names,
        "image_urls": image_urls,
        "ts": ts
    }
    
    # 获取数据库工具实例
    mongo_tool = get_history_mongo_tool()
    
    # 判断是更新还是新增
    if message_id:
        # 更新操作
        mongo_tool.chat_message.update_one(
            {"_id": ObjectId(message_id)},
            {"$set": document}
        )
        return message_id
    else:
        # 新增操作
        result = mongo_tool.chat_message.insert_one(document)
        return str(result.inserted_id)

使用示例：

# 插入用户消息
message_id = save_chat_message(
    session_id="session_001",
    role="user",
    text="你好，请问这个产品怎么用？",
    item_names=["产品A", "产品B"]
)
print(f"插入成功，ID: {message_id}")

# 插入助手回复
save_chat_message(
    session_id="session_001",
    role="assistant",
    text="您好！这个产品的使用方法如下..."
)

二、查询数据（Read）

1. 查询最近 N 条记录

def get_recent_messages(session_id: str, limit: int = 10) -> List[Dict[str, Any]]:
    """
    查询指定会话的最近 N 条对话记录
    :param session_id: 会话唯一标识
    :param limit: 条数限制，默认返回最近 10 条
    :return: 对话记录列表（字典格式）
    """
    mongo_tool = get_history_mongo_tool()
    try:
        # 构造查询条件
        query = {"session_id": session_id}
        
        # 执行查询：按时间戳升序排序，限制返回条数
        cursor = mongo_tool.chat_message.find(query)\
            .sort("ts", ASCENDING)\
            .limit(limit)
        
        # 将游标转为列表
        messages = list(cursor)
        return messages
    except Exception as e:
        logging.error(f"Error getting recent messages: {e}")
        return []

使用示例：

# 查询某个会话的最近 10 条消息
messages = get_recent_messages("session_001", limit=10)
print(f"查询到的记录数: {len(messages)}")

# 遍历打印每条记录
for msg in messages:
    print(f"角色: {msg['role']}")
    print(f"内容: {msg['text']}")
    print(f"时间: {msg['ts']}")
    print("---")

三、更新数据（Update）

1. 批量更新

def update_message_item_names(ids: List[str], item_names: List[str]) -> int:
    """
    批量更新历史会话记录的关联商品名称
    :param ids: 要更新的记录主键 ID 列表
    :param item_names: 要设置的新商品名称列表
    :return: 实际更新的文档数量
    """
    mongo_tool = get_history_mongo_tool()
    try:
        # 将字符串 ID 转为 ObjectId 类型
        object_ids = [ObjectId(i) for i in ids]
        
        # 执行批量更新
        result = mongo_tool.chat_message.update_many(
            {"_id": {"$in": object_ids}},
            {"$set": {"item_names": item_names}}
        )
        
        logging.info(f"Updated {result.modified_count} records")
        return result.modified_count
    except Exception as e:
        logging.error(f"Error updating history item_names: {e}")
        return 0

使用示例：

# 批量更新多条记录的商品名称
ids_to_update = ["64a1b2c3d4e5f6a7b8c9d0e1", "64a1b2c3d4e5f6a7b8c9d0e2"]
updated_count = update_message_item_names(
    ids=ids_to_update,
    item_names=["新产品A", "新产品B", "新产品C"]
)
print(f"更新了 {updated_count} 条记录")

四、删除数据（Delete）

1. 清空指定会话的所有记录

def clear_history(session_id: str) -> int:
    """
    清空指定会话的所有历史对话记录
    :param session_id: 会话唯一标识
    :return: 实际删除的文档数量
    """
    mongo_tool = get_history_mongo_tool()
    try:
        # 执行批量删除操作
        result = mongo_tool.chat_message.delete_many({
            "session_id": session_id
        })
        
        logging.info(f"Deleted {result.deleted_count} messages for session {session_id}")
        return result.deleted_count
    except Exception as e:
        logging.error(f"Error clearing history for session {session_id}: {e}")
        return 0

使用示例：

# 删除某个会话的所有历史记录
deleted_count = clear_history("session_001")
print(f"删除了 {deleted_count} 条记录")

高级特性

1. 索引优化

为提升查询性能，我们在初始化时创建了复合索引：

# 创建复合索引：session_id 升序 + ts 降序
self.chat_message.create_index([("session_id", 1), ("ts", -1)])

索引的作用：

• 加速按 session_id 查询的速度
• 优化按时间戳排序的性能
• 适合"按会话查询最新记录"的场景

2. 错误处理

所有数据库操作都包含了完善的错误处理：

try:
    # 数据库操作
    result = mongo_tool.chat_message.insert_one(document)
except Exception as e:
    logging.error(f"数据库操作失败: {e}")
    # 返回默认值或重新抛出异常
    return None

3. 时间戳记录

每条记录都包含时间戳，方便排序和时间范围查询：

ts = datetime.now().timestamp()
document = {
    "ts": ts,
    # ... 其他字段
}

完整测试示例

if __name__ == "__main__":
    # 测试会话 ID
    sid = "test_session_001"
    
    # 1. 插入数据
    print("=== 插入数据 ===")
    msg_id1 = save_chat_message(sid, "user", "你好")
    msg_id2 = save_chat_message(sid, "assistant", "你好！有什么可以帮助你的？")
    msg_id3 = save_chat_message(sid, "user", "这个产品怎么用？", item_names=["产品A"])
    
    # 2. 查询数据
    print("\n=== 查询数据 ===")
    messages = get_recent_messages(sid, limit=5)
    print(f"查询到的记录数: {len(messages)}")
    for m in messages:
        print(f"角色: {m['role']}, 内容: {m['text']}")
    
    # 3. 更新数据
    print("\n=== 更新数据 ===")
    updated = update_message_item_names([msg_id3], ["产品A", "产品B"])
    print(f"更新了 {updated} 条记录")
    
    # 4. 删除数据
    print("\n=== 删除数据 ===")
    deleted = clear_history(sid)
    print(f"删除了 {deleted} 条记录")
    
    # 5. 验证删除
    print("\n=== 验证删除 ===")
    messages = get_recent_messages(sid, limit=5)
    print(f"删除后查询到的记录数: {len(messages)}")

最佳实践建议

1. 连接管理

• ✅ 使用单例模式避免重复连接
• ✅ 在模块加载时预初始化连接
• ✅ 使用环境变量管理敏感配置

2. 性能优化

• ✅ 为常用查询字段创建索引
• ✅ 使用批量操作替代循环单条操作
• ✅ 合理使用 limit 限制查询结果数量

3. 数据安全

• ✅ 不要硬编码密码和连接字符串
• ✅ 使用 try-except 处理异常
• ✅ 记录详细的操作日志

4. 代码规范

• ✅ 添加完整的类型注解
• ✅ 编写清晰的函数文档字符串
• ✅ 使用有意义的变量名和函数名

常见问题

Q1: 如何处理 ObjectId 转换？

from bson import ObjectId

# 字符串转 ObjectId
obj_id = ObjectId("64a1b2c3d4e5f6a7b8c9d0e1")

# ObjectId 转字符串
str_id = str(obj_id)

Q2: 如何实现分页查询？

def get_messages_with_pagination(session_id: str, page: int = 1, page_size: int = 10):
    """分页查询"""
    skip_count = (page - 1) * page_size
    cursor = mongo_tool.chat_message.find({"session_id": session_id})\
        .sort("ts", ASCENDING)\
        .skip(skip_count)\
        .limit(page_size)
    return list(cursor)

Q3: 如何查询时间范围内的数据？

def get_messages_by_time_range(session_id: str, start_ts: float, end_ts: float):
    """按时间范围查询"""
    query = {
        "session_id": session_id,
        "ts": {"$gte": start_ts, "$lte": end_ts}
    }
    return list(mongo_tool.chat_message.find(query).sort("ts", ASCENDING))

总结

本文通过实际项目代码，详细演示了如何使用 Python 连接 MongoDB 并实现增删改查操作。核心要点包括：

1. 连接配置：使用环境变量管理敏感信息
2. 单例模式：避免重复创建数据库连接
3. 索引优化：提升查询性能
4. 错误处理：保证程序稳定性
5. 类型注解：提高代码可读性

MongoDB 的灵活性使其非常适合存储非结构化或半结构化数据，配合 Python 的强大生态，可以快速构建高性能的数据处理应用。

参考资料

• PyMongo 官方文档
• MongoDB 官方文档
• Python dotenv 文档