66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
import logging
|
||
import mysql.connector.pooling
|
||
import redis
|
||
from typing import Optional
|
||
|
||
|
||
class DBConnectionManager:
|
||
"""数据库连接管理器，负责管理MySQL和Redis连接池"""
|
||
|
||
_instance = None
|
||
|
||
@classmethod
|
||
def get_instance(cls, mysql_config=None, redis_config=None):
|
||
"""获取单例实例
|
||
|
||
Args:
|
||
mysql_config: MySQL配置
|
||
redis_config: Redis配置
|
||
|
||
Returns:
|
||
DBConnectionManager实例
|
||
"""
|
||
if cls._instance is None:
|
||
cls._instance = cls(mysql_config, redis_config)
|
||
elif mysql_config or redis_config:
|
||
# 如果已经有实例但又传入了配置，则更新配置
|
||
if mysql_config and not cls._instance.mysql_pool:
|
||
cls._instance.init_mysql_pool(mysql_config)
|
||
if redis_config and not cls._instance.redis_pool:
|
||
cls._instance.init_redis_pool(redis_config)
|
||
return cls._instance
|
||
|
||
def __init__(self, mysql_config=None, redis_config=None):
|
||
"""初始化数据库连接管理器
|
||
|
||
Args:
|
||
mysql_config: MySQL配置
|
||
redis_config: Redis配置
|
||
"""
|
||
# 防止直接实例化
|
||
if DBConnectionManager._instance is not None:
|
||
return
|
||
|
||
self.mysql_pool = None
|
||
self.redis_pool = None
|
||
|
||
# 初始化MySQL连接池
|
||
if mysql_config:
|
||
self.init_mysql_pool(mysql_config)
|
||
|
||
# 初始化Redis连接池
|
||
if redis_config:
|
||
self.init_redis_pool(redis_config)
|
||
|
||
def get_mysql_connection(self):
|
||
"""获取MySQL连接"""
|
||
if not self.mysql_pool:
|
||
raise Exception("MySQL连接池未初始化")
|
||
return self.mysql_pool.get_connection()
|
||
|
||
def get_redis_connection(self):
|
||
"""获取Redis连接"""
|
||
if not self.redis_pool:
|
||
raise Exception("Redis连接池未初始化")
|
||
return redis.Redis(connection_pool=self.redis_pool) |