51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
import logging
|
||
import mysql.connector.pooling
|
||
import redis
|
||
from typing import Optional
|
||
|
||
|
||
class DBConnectionManager:
|
||
"""数据库连接管理类，用于管理MySQL和Redis连接池"""
|
||
|
||
_instance = None
|
||
|
||
def __new__(cls, *args, **kwargs):
|
||
if cls._instance is None:
|
||
cls._instance = super(DBConnectionManager, cls).__new__(cls)
|
||
cls._instance.initialized = False
|
||
return cls._instance
|
||
|
||
def __init__(self, mysql_config: Optional[dict] = None, redis_config: Optional[dict] = None):
|
||
if self.initialized:
|
||
return
|
||
|
||
self.LOG = logging.getLogger("DBConnectionManager")
|
||
|
||
# 初始化MySQL连接池
|
||
if mysql_config:
|
||
self.mysql_pool = mysql.connector.pooling.MySQLConnectionPool(**mysql_config)
|
||
self.LOG.info(f"MySQL连接池初始化完成: {mysql_config}")
|
||
else:
|
||
self.mysql_pool = None
|
||
|
||
# 初始化Redis连接池
|
||
if redis_config:
|
||
self.redis_pool = redis.ConnectionPool(**redis_config)
|
||
self.LOG.info(f"Redis连接池初始化完成: {redis_config}")
|
||
else:
|
||
self.redis_pool = None
|
||
|
||
self.initialized = True
|
||
|
||
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) |