VberAdminPy/VapAdmin/BACKEND/utils/db_config_loader.py

from urllib.parse import quote_plus
from typing import Dict, Any

from core.settings import db_settings


class DBConfigLoader:
    """
    数据库配置加载器
    负责加载和处理数据库配置，与数据库连接管理分离
    """
    
    @classmethod
    def get_db_config(cls, db_name: str | None = None) -> Dict[str, Any]:
        """
        获取指定数据库的配置
        
        :param db_name: 数据库名称，如果为None则返回默认数据库配置
        :return: 数据库配置字典
        """
        return db_settings.databases[db_name] if db_name else db_settings.databases['default']
    
    @classmethod
    def build_connection_url(cls, config: Dict[str, Any]) -> str:
        """
        根据配置构建数据库连接URL
        
        :param config: 数据库配置字典
        :return: 连接URL字符串
        """
        db_type = config['type']
        driver = 'asyncmy' if db_type == 'mysql' else 'asyncpg'
        
        # 构建数据库连接URL
        url = (
            f"{db_type}+{driver}://{config['username']}:{quote_plus(config['password'])}@"
            f"{config['host']}:{config['port']}/{config['database']}"
        )
        
        return url
    
    @classmethod
    def get_engine_params(cls, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        获取创建引擎所需的参数
        
        :param config: 数据库配置字典
        :return: 引擎参数字典
        """
        return {
            'echo': config['echo'],
            'max_overflow': config['max_overflow'],
            'pool_size': config['pool_size'],
            'pool_recycle': config['pool_recycle'],
            'pool_timeout': config['pool_timeout']
        }