示例代码:
from loguru import logger
from pymongo import MongoClient
from pymongo.errors import ConnectionFailurefrom llm_engineering.settings import settingsclass MongoDatabaseConnector:_instance: MongoClient | None = Nonedef __new__(cls, *args, **kwargs) -> MongoClient:if cls._instance is None:try:cls._instance = MongoClient(settings.DATABASE_HOST)except ConnectionFailure as e:logger.error(f"Couldn't connect to the database: {e!s}")raiselogger.info(f"Connection to MongoDB with URI successful: {settings.DATABASE_HOST}")return cls._instanceconnection = MongoDatabaseConnector()
详细解释
-
导入依赖模块
from loguru import logger- 作用:导入
loguru库中的logger对象。这个对象用于记录日志信息,比标准库中的 logging 更加方便和直观。 - 举例:当程序运行时,可以用
logger.info("提示信息")打印一条信息,比如记录连接成功或失败的日志。
from pymongo import MongoClient- 作用:导入
pymongo库中的MongoClient类。这个类用来创建与 MongoDB 数据库的连接。 - 举例:假如你需要连接一个 MongoDB 数据库,其地址为
"mongodb://127.0.0.1:27017",就可以用MongoClient("mongodb://127.0.0.1:27017")来创建连接对象。
from pymongo.errors import ConnectionFailure- 作用:从
pymongo库中导入ConnectionFailure异常类,当连接数据库失败时,这个异常会被抛出。 - 举例:如果数据库地址错误或者数据库没有启动,会抛出
ConnectionFailure异常,我们可以捕获这个异常并记录错误日志。
from llm_engineering.settings import settings- 作用:导入项目中
llm_engineering.settings模块中的settings对象。这个对象通常存储配置参数,比如数据库的 URI 地址。 - 举例:假设在
settings中有定义DATABASE_HOST = "mongodb://127.0.0.1:27017",那么代码会使用这个地址来连接数据库。
- 作用:导入
-
定义 MongoDatabaseConnector 类
class MongoDatabaseConnector:_instance: MongoClient | None = None- 作用:定义一个类
MongoDatabaseConnector,用来管理 MongoDB 的连接。这里使用了一个类变量_instance来存储连接实例。 - 解释:
_instance初始值为None,表示当前还没有创建数据库连接。类型提示MongoClient | None表示这个变量可能是一个MongoClient对象,也可能是None。
- 作用:定义一个类
-
重写 new 方法实现单例模式
def __new__(cls, *args, **kwargs) -> MongoClient:if cls._instance is None:try:cls._instance = MongoClient(settings.DATABASE_HOST)except ConnectionFailure as e:logger.error(f"Couldn't connect to the database: {e!s}")raiselogger.info(f"Connection to MongoDB with URI successful: {settings.DATABASE_HOST}")return cls._instance-
作用:
__new__是 Python 中的一个特殊方法,用于在创建实例之前进行控制。这里重写它实现了“单例模式”,即无论你创建多少次这个类的实例,都只会返回同一个 MongoClient 连接对象。 -
详细步骤:
-
判断是否已存在连接:
if cls._instance is None:- 意思:如果
_instance还是None(表示还没有连接实例),则继续创建新的连接。
- 意思:如果
-
尝试建立数据库连接:
try:cls._instance = MongoClient(settings.DATABASE_HOST)- 作用:调用
MongoClient构造函数,使用settings.DATABASE_HOST(例如"mongodb://127.0.0.1:27017")来创建数据库连接。 - 举例:假设
settings.DATABASE_HOST定义为"mongodb://192.168.1.100:27017",那么代码会试图连接到 IP 为 192.168.1.100、端口为 27017 的 MongoDB 实例。
- 作用:调用
-
异常处理:
except ConnectionFailure as e:logger.error(f"Couldn't connect to the database: {e!s}")raise- 作用:如果连接失败(例如数据库没有启动、地址错误等),会捕获
ConnectionFailure异常。 - 解释:程序会通过
logger.error输出错误日志,然后使用raise将异常继续抛出,确保调用者知道连接出了问题。 - 举例:若连接失败并抛出异常,比如错误代码为 10061(连接被拒绝),错误日志会显示
"Couldn't connect to the database: [错误信息]"。
- 作用:如果连接失败(例如数据库没有启动、地址错误等),会捕获
-
记录成功日志:
logger.info(f"Connection to MongoDB with URI successful: {settings.DATABASE_HOST}")- 作用:不论是首次连接还是复用已有连接,都记录一条信息日志,表明连接成功。
- 举例:如果成功连接到
"mongodb://192.168.1.100:27017",日志中会显示"Connection to MongoDB with URI successful: mongodb://192.168.1.100:27017"。
-
返回连接实例:
return cls._instance- 作用:返回创建好的 MongoClient 实例。这样,无论何时调用
MongoDatabaseConnector(),最终返回的都是同一个连接实例。
- 作用:返回创建好的 MongoClient 实例。这样,无论何时调用
-
-
-
创建连接实例
connection = MongoDatabaseConnector()- 作用:调用
MongoDatabaseConnector()来获取 MongoDB 的连接实例。 - 解释:这行代码实际上会触发上面定义的
__new__方法,判断是否需要创建新连接或直接返回已有连接。最终,变量connection存储了一个MongoClient对象。 - 举例:如果第一次调用时创建连接,则
connection可能代表一个连接到"mongodb://192.168.1.100:27017"的 MongoClient 实例;后续再调用时将复用这个实例。
- 作用:调用
总结
- 单例模式:该代码使用了单例模式保证整个程序中只会创建一个数据库连接。这样可以避免重复创建连接资源,提升效率。
- 错误处理:使用
try...except捕获连接失败的情况,并通过日志记录错误信息,然后抛出异常,确保问题不会被忽略。 - 日志记录:通过
loguru的logger对象,详细记录了连接的成功与失败状态,方便排查问题。 - 实际场景举例:
假设你的数据库地址为"mongodb://192.168.1.100:27017",第一次执行MongoDatabaseConnector()时:- 程序检查
_instance是None; - 连接成功后,将
MongoClient("mongodb://192.168.1.100:27017")赋值给_instance; - 输出日志 “Connection to MongoDB with URI successful: mongodb://192.168.1.100:27017”;
- 将该连接实例返回并赋值给
connection。
如果之后再调用MongoDatabaseConnector(),程序就直接返回已存在的连接实例,不再重新连接。
- 程序检查
这样写的好处是节省资源、简化连接管理,同时通过日志记录帮助开发者快速定位数据库连接的问题。