common_app_config.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@Email: ludq1@chinaunicom.cn
@date: 2023/04/07 11:40:00
@Description:
"""

import json
import logging
import os
import traceback
from typing import List, Dict, Union

import yaml

from .base_const import ConstBaseApp


class CommonAppConfig:
    r"""
    应用的通用配置类，通过 CommonAppConfig() 获取
    """
    # 通用配置
    is_debug: bool = False
    log_level_dict = {"debug": logging.DEBUG, "info": logging.INFO, "warn": logging.WARN,
                      "warning": logging.WARNING,
                      "error": logging.ERROR, "fatal": logging.FATAL, "critical": logging.CRITICAL}
    # 初始化字典中log_level使用字符串，初始化后转换为对应的log level int值
    log_level: int = None
    runtime_env: str = None

    _common_logger: logging.Logger = None
    r"""通用的debug使用的logger"""

    # 运行时内存缓存
    runtime_cache_dict: dict = None

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, 'inst'):  # 单例
            self = super().__new__(cls, *args, **kwargs)

            # 通过环境变量初始化，应用要连接的数据库、redis等其他配置
            self.init_app_config()
            cls.inst = self
        return getattr(cls, 'inst')

    def init_app_config(self):
        is_debug_str = os.environ.get("IS_DEBUG") or "false"
        self.is_debug = is_debug_str.lower() == "true"

        if self.is_debug:
            self.log_level = logging.DEBUG
        else:
            log_level_str = os.environ.get("LOG_LEVEL") or "info"
            log_level_int = self.log_level_dict.get(log_level_str.lower())
            if log_level_int is None:
                log_level_int = logging.INFO
            self.log_level = log_level_int

        self.runtime_env = os.environ.get("RUNTIME_ENV") or "NOTSET"

    def refresh_app_config(self, conf_filepath: str = ConstBaseApp.COMMON_APP_CONFIG_DEFAULT_CONF_FILE_PATH):
        if not conf_filepath:
            conf_filepath = ConstBaseApp.COMMON_APP_CONFIG_DEFAULT_CONF_FILE_PATH

        try:
            app_config_dict = self.load_file_as_dict(conf_filepath)
        except BaseException as e:
            if conf_filepath != ConstBaseApp.COMMON_APP_CONFIG_DEFAULT_CONF_FILE_PATH:
                self.common_logger.warning(f"读取配置文件{conf_filepath}时发生异常:{e},不刷新CommonAppConfig")
                self.common_logger.warning(traceback.format_exc())
            return

        # 加载默认值
        self.init_app_config()

        for conf_key in {"runtime_env"}:
            conf_value = app_config_dict.get(conf_key)
            if conf_value is not None:
                setattr(self, conf_key, conf_value)

        is_debug = app_config_dict.get("is_debug")
        if is_debug is None:
            # 如果没有配置属性,则什么也不做
            pass
        else:
            if isinstance(is_debug, bool):
                self.is_debug = is_debug
            else:
                is_debug_str = str(is_debug)
                self.is_debug = is_debug_str.lower() == "true"

        if self.is_debug:
            self.log_level = logging.DEBUG
        else:
            log_level = app_config_dict.get("log_level")
            if log_level is None:
                # 如果没有配置属性,则什么也不做
                pass
            else:
                if isinstance(log_level, int):
                    self.log_level = log_level
                else:
                    log_level_str = str(log_level)
                    log_level_int = self.log_level_dict.get(log_level_str.lower())
                    if log_level_int is None:
                        log_level_int = logging.INFO
                    self.log_level = log_level_int

        return

    def load_file_as_dict(self, filepath: str) -> dict:
        r"""
        判断文件后缀名, 如果是 .yaml 或 .yml 则按 yaml 方式读取,
        文件内容不合规,则抛出异常
        """
        with open(filepath, 'r', encoding="utf-8") as dict_file:
            conf_content = dict_file.read()
            # 无论是 yaml 格式 还是 json 格式,都可以使用 yaml_load 读取,但读取的结果可能是 字典或字符串
            dict_content = self.yaml_load(conf_content)
            if isinstance(dict_content, str):
                raise ValueError(f"{filepath}的内容不是有效的json或yaml字符串")

        return dict_content

    def yaml_load(self, yaml_str: str) -> Union[dict, str]:
        r"""
        无论是 yaml 格式 还是 json 格式,都可以使用 yaml_load 读取,但读取的结果可能是 字典或字符串
        """
        dict_list = self.yaml_load_list(cfg_content=yaml_str)
        if dict_list:
            return dict_list[0]
        else:
            return dict()

    def yaml_load_list(self, path: str = None, cfg_content: str = None) -> List[Union[Dict, str]]:
        """
        将yaml格式文件转换为dict值，该yaml文件可以包含多块yaml数据，每个dict放到list中返回,
        无论是 yaml 格式 还是 json 格式,都可以使用 yaml_load 读取,但读取的结果可能是 字典或字符串

        :param path: 文件路径
        :param cfg_content:
        :return: list -> []
        """
        if path:
            cfg = self.read_file_content(file_path=path)
        else:
            cfg = cfg_content or ""

        yaml_generator = yaml.safe_load_all(cfg)  # 将yaml格式文件转换为dict值，该yaml文件可以包含多块yaml数据
        if yaml_generator is None:
            return list()

        # 转成dict并保存到list中
        yaml_list = list()
        for one_yaml_generator in yaml_generator:
            if one_yaml_generator is None:
                continue
            yaml_list.append(json.loads(json.dumps(one_yaml_generator)))
        return yaml_list

    def read_file_content(self, file_path) -> str:
        """
        读取文件内容

        :param file_path: 文件路径
        :return: str -> file-content
        """
        with open(file_path, 'r') as f:  # 打开一个文件，必须是'rb'模式打开
            return f.read()

    @property
    def common_logger(self):
        r"""
        通用的debug使用的logger
        """
        if self._common_logger is None:
            logger = logging.getLogger("COMMON_LOGGER")
            logger.setLevel(self.log_level)
            if not logger.handlers:
                ch = logging.StreamHandler()
                ch.setLevel(logging.DEBUG)
                formatter = logging.Formatter(
                    '%(asctime)s %(name)s - %(pathname)s - func:%(funcName)s - lineno:%(lineno)s - %(levelname)s - %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S'
                )
                ch.setFormatter(formatter)
                logger.addHandler(ch)
            self._common_logger = logger
        return self._common_logger