【MongoDB实战】第12章测试与部署：从开发到生产（python实战完善版）-平芜编程栈

文章目录

《MongoDB实战入门》第12章测试与部署：从开发到生产（python实战完善版）
- 12.1 MongoDB单元测试
- - 12.1.1 环境准备
  - 12.1.2 核心代码实现（覆盖CRUD+聚合+事务）
  - - 第一步：编写待测试的MongoDB业务操作类
    - 第二步：编写单元测试用例
    - 第三步：运行单元测试
- 12.2 生产环境部署最佳实践
- - 12.2.1 环境变量配置（区分开发/测试/生产）
  - - 方案1：本地开发/测试环境（使用.env文件）
    - 方案2：生产环境（系统环境变量配置）
  - 12.2.2 安全配置实操
  - - 1. 开启MongoDB认证模式（服务端配置）
    - 2. 创建权限隔离的MongoDB用户
    - 3. 限制IP访问（防火墙层面）
  - 12.2.3 数据备份与恢复实战（python+Shell）
  - - 1. 定时备份脚本（python实现）
    - 2. 设置定时备份（crontab）
    - 3. 数据恢复脚本（python实现）
    - 4. 恢复命令使用示例
- 12.3 基础监控
- - 12.3.1 MongoDB内置监控工具（mongostat/mongotop）
  - - 1. mongostat：实时监控全局运行状态
    - - 基本使用命令（生产环境带认证）
      - 核心指标解读（重点关注）
    - 2. mongotop：实时监控集合级读写耗时
    - - 基本使用命令（生产环境带认证）
  - 12.3.2 python自定义监控脚本（采集核心指标）
  - - 后台运行监控脚本
- 总结

《MongoDB实战入门》第12章测试与部署：从开发到生产（python实战完善版）

本章聚焦MongoDB从开发测试到生产落地的全流程python实操，补充可直接复用的代码示例，确保每个技术点都能落地执行，核心围绕单元测试、生产部署、监控运维三大模块展开。

12.1 MongoDB单元测试

单元测试的核心目标是脱离真实MongoDB服务依赖，通过模拟环境验证数据操作逻辑的正确性，避免测试污染真实业务数据，python生态中常用pytest（测试框架）+mongomock（MongoDB模拟工具）实现该需求。

12.1.1 环境准备

首先安装所需依赖包：

pipinstallpytest mongomock pymongo python-dotenv

12.1.2 核心代码实现（覆盖CRUD+聚合+事务）

第一步：编写待测试的MongoDB业务操作类

创建mongodb_service.py，封装核心业务操作：

# mongodb_service.pyfrompymongoimportMongoClientfromdatetimeimportdatetimefromtypingimportList,Dict,Optional,UnionclassMongoDBBusinessService:"""MongoDB业务操作封装类，包含CRUD、聚合、事务等核心操作"""def__init__(self,mongo_client:MongoClient,db_name:str="business_db"):self.db=mongo_client[db_name]self.collection=self.db["user_orders"]# 订单业务集合# 新增数据（C）：创建订单defcreate_order(self,order_info:Dict)->str:""" 创建订单 :param order_info: 订单信息字典，必须包含user_id、order_amount字段 :return: 新增订单的ID字符串 :raises ValueError: 缺少必填字段时抛出异常 """required_fields=["user_id","order_amount"]ifnotall(fieldinorder_infoforfieldinrequired_fields):raiseValueError("订单信息缺少必填字段（user_id/order_amount）")# 补充默认字段order_info.update({"create_time":datetime.now(),"order_status":"pending","update_time":datetime.now()})result=self.collection.insert_one(order_info)returnstr(result.inserted_id)# 查询数据（R）：按用户ID查询订单defget_orders_by_user(self,user_id:str,limit:int=10)->List[Dict]:""" 按用户ID查询订单列表 :param user_id: 用户唯一标识 :param limit: 返回订单数量上限 :return: 订单列表（去除_id字段） """query={"user_id":user_id}projection={"_id":0}# 不返回_id字段orders=list(self.collection.find(query,projection).limit(limit).sort("create_time",-1))returnorders# 更新数据（U）：更新订单状态defupdate_order_status(self,order_id:str,new_status:str)->int:""" 更新订单状态 :param order_id: 订单ID :param new_status: 新状态（pending/paid/shipped/completed） :return: 受影响的记录数 """query={"_id":order_id}update_data={"$set":{"order_status":new_status,"update_time":datetime.now()}}result=self.collection.update_one(query,update_data)returnresult.modified_count# 删除数据（D）：删除过期无效订单defdelete_invalid_orders(self,user_id:str)->int:""" 删除用户的无效订单（状态为pending且创建时间超过24小时，仅示例） :param user_id: 用户唯一标识 :return: 删除的记录数 """# 模拟24小时前的时间（实际可根据业务调整）expire_time=datetime.now()-timedelta(hours=24)query={"user_id":user_id,"order_status":"pending","create_time":{"$lt":expire_time}}result=self.collection.delete_many(query)returnresult.deleted_count# 聚合查询：按用户统计订单总金额defstat_user_order_total(self)->List[Dict]:""" 聚合查询：统计每个用户的订单总金额、订单数量 :return: 聚合结果列表 """pipeline=[{"$group":{"_id":"$user_id","total_amount":{"$sum":"$order_amount"},"order_count":{"$sum":1}}},{"$sort":{"total_amount":-1}},{"$limit":20# 仅返回前20名用户}]returnlist(self.collection.aggregate(pipeline))# 事务操作：批量创建订单（需MongoDB 4.0+支持，mongomock兼容事务模拟）defbatch_create_orders(self,order_list:List[Dict])->bool:""" 事务批量创建订单，要么全部成功，要么全部回滚 :param order_list: 订单列表 :return: 批量创建是否成功 """ifnotorder_list:returnFalsewithself.db.client.start_session()assession:session.start_transaction()try:fororderinorder_list:self.create_order(order)session.commit_transaction()returnTrueexceptExceptionase:session.abort_transaction()raiseException(f"批量创建订单失败：{str(e)}")

第二步：编写单元测试用例

创建test_mongodb_service.py，使用pytest+mongomock实现无依赖测试：

# test_mongodb_service.pyimportpytestimportmongomockfrompymongoimportMongoClientfromdatetimeimporttimedelta,datetimefrommongodb_serviceimportMongoDBBusinessService# 测试夹具1：创建模拟MongoDB客户端（无需真实服务，全局复用）@pytest.fixture(scope="function")defmock_mongo_client():"""创建模拟MongoDB客户端，测试完成后自动清理数据"""client=mongomock.MongoClient()# 完全兼容pymongo APIyieldclient# 测试结束后删除测试数据库，避免数据残留client.drop_database("business_db")# 测试夹具2：初始化业务服务实例@pytest.fixture(scope="function")deforder_service(mock_mongo_client):"""初始化MongoDB业务服务，依赖模拟客户端"""returnMongoDBBusinessService(mock_mongo_client)# -------------- 测试用例：覆盖CRUD核心操作 --------------deftest_create_order_success(order_service):"""测试创建订单（正常场景）"""order_data={"use