Как изменить формат logger.info(...) в str или передать в переменную(python)?

Question

[Robert]

Здраствуйте. Я создал плюс минус работающий logger для приложения fastapi. Даные выводяться в консоль через logger.info(...) и logger.error(...). Хотел бы сохранить эти ошибки в бд. А когда пытаюсь передать loger.info(...) в переменую выдает: ошибка типа. Тип None.
Как передать это в переменную для записи в бд? спасибо
P.C не судите строго

Answer 1

[Vindicar]

Никак, логгеры не возвращают сообщения об ошибках.
Пиши класс-потомок logger.Handler который будет сохранять сообщения в БД, и назначай экземпляр этого хандлера нужным логам.
Подробности в доках на модуль logging, но если кратко - там разделение обязанностей. Есть Formatter, который занимается превращением записи в строку, есть Handler, который пихает полученную строку в файл или ещё куда, есть Filter, который реализует умную фильтрацию записей, ну и т.д.

Comments

[Robert]

понял, спасибо

Answer 2

[kamenyuga]

Вот пример как можно реализовать хранение логов в бд оракл.

код

# built-in modules
import datetime
import functools
import logging
import pathlib
import time

# third-party modules
import cx_Oracle as cxO

# project modules
import utilities_config as cutil


class DBHandler(logging.Handler):

    """
    Класс-обработчик логирования в базу данных (оракл)
    """

    _query_create = f"""
        create table {cutil.LOGFIG_STORAGE_LOCATION['log_table']} (
            logger_name varchar2(100 char),
            logging_started date,
            idx number,
            record varchar2(1000 char))"""
    _query_insert = f"""
        insert into {cutil.LOGFIG_STORAGE_LOCATION['log_table']}
        values (:1, :2, :3, :4)"""

    def __init__(self, logger_name, logging_started, chunk_size=100):

        super().__init__()

        self.logger_name = logger_name
        self.logging_started = logging_started

        self.chunk_size = chunk_size
        self.count = 0
        self.chunk = list()
        self._create()

    def emit(self, record):
        """
        Обработка каждого лог-сообщения
        """
        self.count += 1
        self.chunk.append((self.logger_name, self.logging_started, self.count, self.format(record)))
        if len(self.chunk) > 0 and len(self.chunk) % self.chunk_size == 0:
            self.flush()

    def flush(self):
        """
        Запись накопившихся лог-сообщений
        """
        try:
            self._insert()
        except cxO.DatabaseError as exc:
            logging.getLogger(self.logger_name).error(f"can't flush logs to oracle: {exc}")
        self.chunk.clear()

    def _create(self):
        """
        Подключение к базе и создание таблицы
        """
        with cxO.connect(**cutil.LOGFIG_ORACLE_CONNECTION) as connection:
            cursor = connection.cursor()
            try:
                cursor.execute(self._query_create)
            except cxO.DatabaseError as exc:
                # пропускаем ошибку
                # ORA-00955: name is already used by an existing object
                # так как в оракле нет синтаксиса 'CREATE TABLE IF NOT EXISTS ...'
                if exc.args[0].code != 955:
                    # все остальные ошибки просто пробрасываем выше
                    raise exc

    def _insert(self):
        """
        Подключение к базе и вставка строк
        """
        with cxO.connect(**cutil.LOGFIG_ORACLE_CONNECTION) as connection:
            cursor = connection.cursor()
            cursor.executemany(self._query_insert, self.chunk)
            connection.commit()


def get_logger(logger_name, *, log_to_file, log_to_oracle):

    """
    Создание логгера с заданным именем
    """

    # общий для всех шаблон формата лог-сообщений
    fmt = ('[%(asctime)s] [%(levelname)s] %(message)s', '%Y.%m.%d %H:%M:%S')

    # создаем логгер
    logging_started = datetime.datetime.now()
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.DEBUG)

    # [1/3] - логирование в консоль - обработчик и формат лог-сообщений
    ch = logging.StreamHandler()
    ch.setFormatter(logging.Formatter(*fmt))
    logger.addHandler(ch)

    # [2/3] - логирование в файл - обработчик и формат лог-сообщений
    if log_to_file:
        # папка с логами
        log_dir = pathlib.Path(cutil.LOGFIG_STORAGE_LOCATION['log_dir'])
        log_dir.mkdir(exist_ok=True)
        # логгер
        fh = logging.FileHandler(
            pathlib.Path(
                log_dir,
                f"{logger_name}_log_{logging_started.strftime('%Y.%m.%d_%H.%M.%S')}.log"),
            # вот здесь кодировка имеет значение, иначе будет системная кодировка (windows-1251)
            encoding='utf-8')
        fh.setFormatter(logging.Formatter(*fmt))
        logger.addHandler(fh)

    # [3/3] - логирование в оракл - обработчик и формат лог-сообщений
    # оборачиваем в try/except, так как логирование в оракл опционально
    if log_to_oracle:
        try:
            oh = DBHandler(logger_name, logging_started)
        except cxO.DatabaseError as exc:
            logger.error(f"can't create logger to oracle: {exc}")
        else:
            oh.setFormatter(logging.Formatter(*fmt))
            logger.addHandler(oh)

    return logger