MongoDB CRUD类及Jupyter测试脚本问题排查求助

一、初始化方法（__init__）核心问题 #

• 连接验证无效：if not self.client: 无法判断MongoDB是否真的连接成功，MongoClient对象即便连接失败也不会返回False，必须通过self.client.admin.command('ping')主动验证连通性。
• 敏感信息硬编码：USER、PASS等密钥直接写在代码中，极易泄露，应改用环境变量或配置文件传递。
• 冗余字符串格式化：self.client['%s' % (DB)] 可简化为self.client[DB]，字典直接通过键名访问即可。

二、Create方法问题 #

• 未捕获插入异常：insert_one可能触发连接中断、重复键冲突等错误，当前代码未处理这些异常，会直接导致程序崩溃，需添加try-except块捕获并处理。

三、Read方法问题 #

• 空查询逻辑风险：若传入query=None，find(query)会返回集合所有文档，可能引发意外的大量数据返回，需默认设置空字典query = query or {}，避免无意义的全量查询。

四、Jupyter Notebook冗余问题 #

• 重复配置无效：Notebook中再次定义的USER、PASS等参数，被AnimalShelter类__init__里的硬编码值覆盖，完全无效。必须修改类的初始化逻辑，支持外部传入配置参数。

修改后的AnimalShelter类代码 #

from pymongo import MongoClient
from bson.objectid import ObjectId
import os

class AnimalShelter(object):

    def __init__(self, user=None, password=None, host=None, port=None, db='aac', col='animals'):
        # 优先级：传入参数 > 环境变量 > 默认值
        self.USER = user or os.getenv('MONGO_USER')
        self.PASS = password or os.getenv('MONGO_PASS')
        self.HOST = host or os.getenv('MONGO_HOST', 'localhost')
        self.PORT = port or int(os.getenv('MONGO_PORT', 27017))
        self.DB = db
        self.COL = col

        try:
            self.client = MongoClient(
                f'mongodb://{self.USER}:{self.PASS}@{self.HOST}:{self.PORT}',
                server_selection_timeout=60
            )
            # 主动验证连接
            self.client.admin.command('ping')
            self.database = self.client[self.DB]
            self.collection = self.database[self.COL]
        except Exception as e:
            raise Exception(f"MongoDB连接失败: {str(e)}")

    def create(self, data):
        if data is None:
            raise ValueError("数据参数为空，无法保存")
        try:
            result = self.collection.insert_one(data)
            return result.inserted_id is not None
        except Exception as e:
            print(f"插入数据失败: {str(e)}")
            return False

    def read(self, query=None):
        query = query or {}
        try:
            cursor = self.collection.find(query)
            return list(cursor)
        except Exception as e:
            print(f"读取数据失败: {str(e)}")
            return []

修改后的Jupyter Notebook代码 #

from CRUD_Modules import AnimalShelter  

# 通过参数传入配置（或改用环境变量）
animal_shelter = AnimalShelter(
    user='private',
    password='private',
    host='private',
    port=private
)

# 测试数据
data_to_insert = {
    'name': 'Shadow',
    'species': 'Dog',
    'age': 3
}

# 测试创建
create_result = animal_shelter.create(data_to_insert)
print(f"创建结果: {create_result}")

# 测试读取
query_to_read = {'name': 'Shadow'}
read_result = animal_shelter.read(query_to_read)
print(f"读取结果: {read_result}")

内容的提问来源于stack exchange，提问作者riuthu mathews