src/serv/log_init.py - master - Ulyanoff/mlops1

src/serv/log_init.py
70 строк4 KB
Artem Ivochkin
развитие тестирования
15 дек 2025, 13:35
15 дек 2025, 13:3560bb4af
# -*- coding: utf-8 -*-
import logging
import platform
import sys

from serv.consts import color
from serv.req_id import ctx_req_id


class CustomFormatter(logging.Formatter):
    """Модифицирует поведение стандартного форматтера, от которого и наследуется"""

    def format(self, record):
        """Красит букву уровня логирования и удаляет переносы строк в логировании Алхимии"""

        if 'sqlalchemy.engine' in record.name:
            record.msg = record.msg.replace('\n', '').replace('\t', ' ').replace('\r', '')

        formatted_record = super().format(record)  # стандартный формат
        pos = 20 if 0 < self._fmt.find('asctime') < 5 else 0  # включено или отключено логирование времени
        painted_letter = color[record.levelno] + formatted_record[pos] + color[0]
        painted_record = formatted_record[:pos] + painted_letter + formatted_record[pos + 1:]

        return painted_record


class InjectingReqID(logging.Filter):
    """Фильтр, добавляющий номер запроса из переменной контекста в строки логов"""
    def filter(self, record) -> bool:
        """Record - объект строки логирования"""
        try:
            record.req_id = ctx_req_id.get()  # этот номер сгенерирован во время асинхронного старта обработки запроса
        except LookupError:
            record.req_id = "000000-000"
        return True


def setup_log(cfg) -> None:
    """Установка общих параметров логирования"""
    boot = logging.getLogger("boot")

    #: Установка потокового обработчика handler для стандартного вывода stderr
    stdout_handler = logging.StreamHandler(sys.stderr)
    stdout_handler.setFormatter(CustomFormatter(cfg.log_format, "%m/%d %H:%M:%S"))

    #: Добавление фильтра, который ничего не фильтрует, но актуализирует req_id из контекста
    add_req_id = InjectingReqID()
    stdout_handler.addFilter(add_req_id)

    #: ВАЖНО! если это вдруг не первая установка конфигурации логирования, то нужно передать keyword force=True
    logging.basicConfig(level=cfg.log_int, handlers=[stdout_handler], datefmt='%m/%d %H:%M:%S', force=True)

    #: получим список всех логеров в системе
    ar = sorted(list(set([logger_name for logger_name in logging.Logger.manager.loggerDict.keys()])))
    boot.info(f"{len(ar)} loggers created")

    #: Логирование файловой системы
    os_platform = platform.system()
    if os_platform == "Windows":
        boot.info(f"OS {os_platform}, fcntl_win used (files descriptor control system)")
    else:
        boot.info(f"OS {os_platform}, fcntl used (files descriptor control system)")

    #: пространство имён логирования Алхимии
    logging.getLogger('sqlalchemy.engine').setLevel(cfg.sqlalchemy.log.engine_int)      # SQL запросы
    logging.getLogger('sqlalchemy.pool').setLevel(cfg.sqlalchemy.log.pool_int)          # Пулы соединений
    logging.getLogger('sqlalchemy.dialects').setLevel(cfg.sqlalchemy.log.dialects_int)  # Диалекты
    logging.getLogger('sqlalchemy.orm').setLevel(cfg.sqlalchemy.log.orm_int)            # ORM события

    boot.info('Logging initialized')