Python, 原创,

python自定义logging 用阿里云日志或本地文件，分等级保存至不同文件实现

python的logging模块会把几种等级的log都存在同一文件夹下，不容易排错，下面我们把log根据不同等级存储到不同的文件中。

重点：

基于本地文件存储的log，为防止相关level水平文件会记录该level以上的log，需要使用函数 logger = logging.getLogger(str(level)) 如果不带此参数则会把高于当前level的一并存储。

另外，获取记录日志的函数、行号和文件，我们可能需要使用到 sys._getframe().f_back，甚至 sys._getframe().f_back.f_back.

文件目录：

main.py
config/
— development.yaml
— production.yaml
— config.yaml
helpers/
— config_helper.py
— logger_helper.py
— __init__.py
logs/

具体实现：首先我们读取出log的配置信息，读取配置信息的方法可以参考 python读取yaml配置 包含其他文件include 包含配置项拼装join , log的配置在yaml中，该篇文章基于上篇的文件目录结构等信息。LogDriver为aliyun时会使用到logAliyun的参数配置。具体参数如下, development.yaml

# 该配置在yaml中的结构，具体参考上面提到的blog读取
# LOG
Log:
  LogDriver: file # file, aliyun
  LogFileExtension: log
  LogPath: logs/
  # log level
  #    'CRITICAL': CRITICAL,
  #    'FATAL': FATAL,
  #    'ERROR': ERROR,
  #    'WARN': WARNING,
  #    'WARNING': WARNING,
  #    'INFO': INFO,
  #    'DEBUG': DEBUG,
  #    'NOTSET': NOTSET,
  LogLevel: DEBUG
  LogFileLevel: WARNING
  LogAliyun:
    EndPoint: cn-shanghai.log.aliyuncs.com  # 创建Project所属区域匹配的Endpoint
    Project: fc-test-service
    LogStore: fc-test-qa
    AccessKeyId: LTAIPJkrTFYHOcaC
    AccessKeySecret: Di8T1ODTfS2TdYsuoEK3SaBQYYQk7g

python需要引用的模块有 aliyun.log, 需要安装模块 aliyun-log-python-sdk （pip install aliyun-log-python-sdk）。 helpers/logger_helper.py文件代码如下

import logging
import os.path
import time
import sys
from . import config_helper
from aliyun.log import *

log_config = config_helper.get_config("Log")


def get_init_logger(level: str = 'info'):
    try:
        # 获取log配置
        if log_config['LogDriver'].lower() == 'aliyun':
            return log_handdle_aliyun(log_config)
        else:
            return log_handdle_file(level, log_config)
    except Exception as ex:
        print(ex)


def log_handdle_aliyun(config: any):
    return LogAliyun(config)


def log_handdle_file(level: str, config: any):
    # 第一步，创建一个logger
    logger = logging.getLogger(str(level))
    logger.setLevel(config["LogLevel"])  # Log等级总开关
    # 第二步，创建一个handler，用于写入日志文件
    rq = time.strftime('%Y-%m-%d', time.localtime(time.time()))
    current_file_path = __file__
    log_path = os.path.abspath(
        os.path.join(current_file_path, os.pardir, os.pardir, config["LogPath"]))
    log_name = '{}_{}.{}'.format(level, rq, config["LogFileExtension"])
    logfile = log_path + '/' + log_name
    fh = logging.FileHandler(logfile)
    fh.setLevel(config["LogFileLevel"])  # 输出到file的log等级的开关
    # 第三步，定义handler的输出格式
    # formatter = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")
    formatter = logging.Formatter("[%(asctime)s] [%(process)d] [%(levelname)s] [%(mFuncName)s] -"
                                  " (%(mFilename)s:%(mLineno)d) - %(message)s")
    if level == 'error':
        formatter = logging.Formatter("[%(asctime)s] [%(process)d] [%(levelname)s] "
                                      "(%(mPathname)s:%(mLineno)d) - %(message)s")
    if level == 'sql':
        formatter = logging.Formatter("[%(asctime)s] [%(mFuncName)s] -"
                                      " (%(mFilename)s:%(mLineno)d)  [SQL] - %(message)s")
    fh.setFormatter(formatter)
    # 第四步，将logger添加到handler里面
    logger.addHandler(fh)
    return logger


class LogAliyun:
    client: any
    config: any

    def __init__(self, config: any):
        aliyun_config = config['LogAliyun']
        client = LogClient(aliyun_config['EndPoint'], aliyun_config['AccessKeyId'], aliyun_config['AccessKeySecret'])
        self.client = client
        self.config = aliyun_config

    def log(self, level: str, message: str, extra=None):
        topic = ""
        source = ""
        log_item_list = []  # LogItem list
        # contents = [('msg', message), ('level', level), ('functionName', 'oap')]
        contents = [('msg', message), ('level', level), ('functionName', 'oap'), ('extra', extra)]
        log_item = LogItem()
        log_item.set_time(int(time.time()))
        log_item.set_contents(contents)
        log_item_list.append(log_item)
        try:
            req2 = PutLogsRequest(self.config['Project'], self.config['LogStore'], topic, source, log_item_list)
            self.client.put_logs(req2)
        except TypeError as e:
            print(repr(e) + " : " + message)

    def debug(self, message, extra):
        self.log("debug", message, extra)

    def info(self, message, extra):
        self.log("info", message, extra)

    def warning(self, message, extra):
        self.log("warning", message, extra)

    def error(self, message, extra):
        self.log("error", message, extra)

    def critical(self, message, extra):
        self.log("error", message, extra)


# 单例模式调用log
class LogSingle:
    __instance = {}

    @classmethod
    def get_instance(cls, level: str):
        if len(cls.__instance) > 0:
            if level in cls.__instance:
                return cls.__instance[level]
        cls.__instance[level] = get_init_logger(level)
        return cls.__instance[level]


#此函数拥有获取打印执行log的地方，f_back为上一层执行代码的信息，由于我们自己封装了相关函数，所有这里需要获取到上一层
def get_log_run_file_extra():
    log_file = sys._getframe().f_back.f_back
    return {"mFuncName": log_file.f_code.co_name, "mPathname": log_file.f_code.co_filename,
            "mLineno": log_file.f_lineno, "mFilename": os.path.basename(log_file.f_code.co_filename)}


def notset(message):
    extra = get_log_run_file_extra()
    log_handle = LogSingle.get_instance("info")
    log_handle.info(message, extra=extra)


def debug(message):
    extra = get_log_run_file_extra()
    log_handle = LogSingle.get_instance("debug")
    log_handle.debug(message, extra=extra)


def info(message):
    extra = get_log_run_file_extra()
    log_handle = LogSingle.get_instance("info")
    log_handle.info(message, extra=extra)


def warning(message):
    extra = get_log_run_file_extra()
    log_handle = LogSingle.get_instance("warning")
    log_handle.warning(message, extra=extra)


def warn(message):
    extra = get_log_run_file_extra()
    log_handle = LogSingle.get_instance("warning")
    log_handle.warning(message, extra=extra)


def error(message):
    extra = get_log_run_file_extra()
    log_handle = LogSingle.get_instance("error")
    log_handle.error(message, extra=extra)


def fatal(message):
    extra = get_log_run_file_extra()
    log_handle = LogSingle.get_instance("error")
    log_handle.error(message, extra=extra)


def critical(message):
    extra = get_log_run_file_extra()
    log_handle = LogSingle.get_instance("error")
    log_handle.critical(message, extra=extra)

调用代码如下：

from helpers import logger_helper

#调用
logger_helper.info("info")
logger_helper.error("error")
logger_helper.debug("debug")

logs目录下会生成如 error_2020-09-19.log 格式文件。 内容格式如下：

[2020-09-19 11:25:34,810] [6000] [ERROR] (E:\development\python\FrameFastapi\repositories\fmrs_model_repository.py:64) - (pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on 'drdshbga2vouv05v.drds.aliyuncs.com' (timed out)")

(239)

Related Post