WTOAAM/dataAnalysisBehavior/common/appConfig.py

import os
from utils.jsonUtil import JsonUtil
from dataclasses import dataclass
from utils.logUtil import LogUtil
from utils.rdbmsUtil.databaseUtil import DatabaseUtil
from utils.minioUtil.minioClientPool import MinioClientPool
from utils.minioUtil.threadSafeMinioClient import ThreadSafeMinioClient
from utils.config_loader import ConfigLoader
from algorithmContract.const import *
import logging

def GetLogger():
    """返回配置好的日志器"""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler('log/app.log'),
            logging.StreamHandler()
        ]
    )
    return logging.getLogger(__name__)

@dataclass
class DatabaseConfig:
    type: str
    url: str
    timeout: int  # Timeout in seconds
    poolSize: int
    maxPoolSize: int
    minPoolSize: int
    maxIdleTime: int  # Max idle time in seconds


@dataclass
class MinioConfig:
    endpoint: str
    accessKey: str
    secretKey: str
    secure: bool
    poolSize: int
    timeout: int  # Timeout in seconds


@dataclass
class LoggingConfig:
    logFilePath: str
    maxFileSize: int
    maxTotalSize: int
    backupCount: int
    logFormat: str
    level: str


@dataclass
class AppConfig:
    databaseConfig: dict
    minioConfig: MinioConfig
    loggingConfig: LoggingConfig

    @staticmethod
    def loadFromFile(filePath: str):
        configJson = JsonUtil.read_json(filePath)
        businessFoundationDb = DatabaseConfig(
            **configJson['databaseConfig'][DATABASE_BusinessFoundationDb])
        businessDb = DatabaseConfig(
            **configJson['databaseConfig'][DATABASE_businessDb])
        minio = MinioConfig(**configJson['minioConfig'])
        logging = LoggingConfig(**configJson['loggingConfig'])
        return AppConfig({
            "businessFoundationDb": businessFoundationDb,
            "businessDb": businessDb
        }, minio, logging)


# Example usage
configFilePath = os.environ.get('APP_CONFIG_FILE', r"conf/appConfig.json")
config = AppConfig.loadFromFile(configFilePath)
# config = AppConfig.loadFromFile(r'conf/appConfig.json')


def GetLogger():
    return LogUtil(config.loggingConfig.logFilePath, config.loggingConfig.logFormat,
                   config.loggingConfig.maxTotalSize, config.loggingConfig.backupCount).getLogger()


def GetBusinessFoundationDbUtil():
    return DatabaseUtil(url=config.databaseConfig[DATABASE_BusinessFoundationDb].url,
                        pool_size=config.databaseConfig[DATABASE_BusinessFoundationDb].poolSize,
                        max_overflow=config.databaseConfig[DATABASE_BusinessFoundationDb].maxPoolSize,
                        connect_timeout=config.databaseConfig[DATABASE_BusinessFoundationDb].timeout)


def GetBusinessDbUtil():
    return DatabaseUtil(url=config.databaseConfig[DATABASE_businessDb].url,
                        pool_size=config.databaseConfig[DATABASE_businessDb].poolSize,
                        max_overflow=config.databaseConfig[DATABASE_businessDb].maxPoolSize,
                        connect_timeout=config.databaseConfig[DATABASE_businessDb].timeout)


def GetDbUtil():
    dbInstance = {}
    return {
        DATABASE_BusinessFoundationDb: DatabaseUtil(url=config.databaseConfig[DATABASE_BusinessFoundationDb].url,
                                                    pool_size=config.databaseConfig[DATABASE_BusinessFoundationDb].poolSize,
                                                    max_overflow=config.databaseConfig[
                                                        DATABASE_BusinessFoundationDb].maxPoolSize,
                                                    connect_timeout=config.databaseConfig[DATABASE_BusinessFoundationDb].timeout),
        DATABASE_businessDb: DatabaseUtil(url=config.databaseConfig[DATABASE_businessDb].url,
                                          pool_size=config.databaseConfig[DATABASE_businessDb].poolSize,
                                          max_overflow=config.databaseConfig[DATABASE_businessDb].maxPoolSize,
                                          connect_timeout=config.databaseConfig[DATABASE_businessDb].timeout)
    }


def GetMinIOUtil():
    pool = MinioClientPool(config.minioConfig.endpoint, config.minioConfig.accessKey,
                           config.minioConfig.secretKey, config.minioConfig.poolSize)
    return ThreadSafeMinioClient(pool)