Создание пользовательских обработчиков для логов в Python

Представьте ситуацию: ваше приложение работает в продакшене, как вдруг происходит критическая ошибка. Вы узнаете о ней только через несколько часов, когда пользователи начинают массово жаловаться. Идёте проверять консоль, а тут всего лишь трассировка стека, которая мало что говорит о проблеме. Из-за кого и когда возникла это ошибка? Чтобы предотвратить такие сценарии, необходима активная система уведомлений. В этом руководстве мы создадим пользовательский обработчик, который предоставит возможность создания системы уведомлений об ошибках, которая гарантирует, что вы всегда будете в курсе состояния вашей системы.

Теоретическая основа

Модуль logging ^[1] реализует иерархическую систему логирования, которая состоит из 4-х ключевых компонентов:

• Logger ^[2] — ключевой объект для работы с логами. Имеет имя и уровень логирования (DEBUG, INFO, WARNING, ERROR, CRITICAL).

• Handler ^[3] — определяет, как и куда отправлять сообщения (файл, консоль, сокет, email и т.д.). Один логгер может иметь несколько обработчиков.

• Filter ^[4] — используется для дополнительной фильтрации сообщений (например, пропускать только определённые уровни или сообщения с конкретным текстом).

• Formatter ^[5] — определяет, как будет выглядеть сообщение (дата, уровень, модуль, текст и т.д.).

Handler ^[6] — это базовый класс для всех обработчиков логов, в котором некоторые методы определяются только в его потомках. Он отвечает за отправку сообщений из логгера в конечное место назначения.
StreamHandler ^[7] — это тот самый потомок, предназначенный для логирования ошибок в консоль, его мы и будем его переопределять.
LogRecord ^[8] — структура данных, которая содержит всю информацию о конкретном событии логирования.

Основные методы класса Handler:

• format ^[9] — Преобразует объект LogRecord в строку и возвращает её в отформатированном виде.

• emit ^[10] — Определяет, как именно сообщение будет выведено (например, вывод в консоль, запись в файл).

• handle ^[11] — Проверяет фильтры и уровень записи, вызывает emit, если запись прошла проверки.

Чего мы хотим?

Для начала нам следует определить, что мы хотим от нашего пользовательского обработчика. Выделю несколько пунктов:

1. Получать достаточную информацию в консоли, отформатированная в читаемый вид, для определения и устранения проблем и отладки приложения (например, время, модуль, уровень и текст ошибки)

2. Иметь возможность получать уведомления в зависимости от уровня события (например, стандартные ошибки записывать в специальный файл, а при критических ошибках отправлять уведомление разработчику)

Пишем код

Начнём с определения нового класса и его конструктора. Он будет унаследован от StreamHandler. Для реализации 2-го пункта нам нужны callback-функции, которые будут вызываться при обработке события определённого уровня. Сохраним их в словаре, где ключом будет уровень логирования.

class MyHandler(StreamHandler):
    def __init__(
        self, 
        on_critical: Optional[Callable[[LogRecord], None]] = None, 
        on_error: Optional[Callable[[LogRecord], None]] = None, 
        on_warning: Optional[Callable[[LogRecord], None]] = None, 
        on_info: Optional[Callable[[LogRecord], None]] = None,
        on_debug: Optional[Callable[[LogRecord], None]] = None
    ):
        super().__init__()
        self.callbacks = {
            logging.CRITICAL: on_critical,
            logging.ERROR: on_error,
            logging.WARNING: on_warning,
            logging.INFO: on_info,
            logging.DEBUG: on_debug,
        }

Перед тем как вывести текст, нам следует его отформатировать для вывода. Сделаем так, что для каждого уровня события будет применять особый формат.

def format(self, record: LogRecord): 
    log_fmt = LEVEL_FORMATS.get(record.levelno, LOG_FORMAT)
    formatter = Formatter(log_fmt, datefmt=TIME_FORMAT)
    return formatter.format(record)

Реализацию констант, приведённых выше, я вам предлагаю реализовать самостоятельно. Атрибуты для форматирования текста ^[12] и временных меток ^[12] вам в помощь. Ниже я приведу пример из своего приложения:

GREY = "x1b[38;20m"
YELLOW = "x1b[33;20m"
RED = "x1b[31;20m"
BOLD_RED = "x1b[31;1m"
RESET = "x1b[0m"

TIME_FORMAT = '%Y-%m-%d %H:%M:%S'
LOG_FORMAT = "[%(asctime)s %(name)s %(levelname)s %(filename)s:%(lineno)d/%(funcName)s] %(message)s"

LEVEL_FORMATS = {
    logging.DEBUG: GREY + LOG_FORMAT + RESET,
    logging.INFO: GREY + LOG_FORMAT + RESET,
    logging.WARNING: YELLOW + LOG_FORMAT + RESET,
    logging.ERROR: RED + LOG_FORMAT + RESET,
    logging.CRITICAL: BOLD_RED + LOG_FORMAT + RESET,
}

Приступим теперь к логике нашего обработчика. Для этого копируем реализацию метода emit из родительского класса и модифицируем его под наши нужды. Для вывода информации в своём проекте я использую tqdm ^[13], чтобы полоса загрузки в логах отображалась снизу. Вы же можете использовать любой из доступных способов вывода информации в консоль. Так же тут мы и вызываем нужный нам метод при обработке события.

def emit(self, record: LogRecord):
    try:
        msg = self.format(record)
        tqdm.write(msg, end=self.terminator)
        callback = self.callbacks.get(record.levelno)
        if callback:
            callback(record)
    except RecursionError:
        raise
    except Exception as e:
        self.handleError(record)

Теперь нам нужно написать метод для его установки. В аргументы пойдёт всё нужное для класса Logger и методы, которые будут вызваны при определённом уровне события.

def setup_logger(
        name: str,
        level: int = logging.INFO,
        on_critical: Optional[Callable[[LogRecord], None]] = None,
        on_error: Optional[Callable[[LogRecord], None]] = None,
        on_warning: Optional[Callable[[LogRecord], None]] = None,
        on_info: Optional[Callable[[LogRecord], None]] = None,
        on_debug: Optional[Callable[[LogRecord], None]] = None,
    ) -> logging.Logger:
    logger = logging.getLogger(name)
    logger.setLevel(level)

    if not logger.handlers:
        handler = MyHandler(
            on_critical=on_critical,
            on_error=on_error,
            on_warning=on_warning,
            on_info=on_info,
            on_debug=on_debug,
        )
        logger.addHandler(handler)

    return logger

Обработчик готов! Теперь его можно установить куда угодно, где вы хотите отслеживать события. Методы on_error, on_critical и прочие пусть будет реализованы на ваше усмотрение. Для проверки их работоспособности просто выведем текст.

def on_error(record: logging.LogRecord):
    print("Возникла ошибка")
    # Реалзация

def on_critical(record: logging.LogRecord):
    print("Возникла критическая ошибка")
    # Реалзация

mylogger = setup_logger(__name__, logging.INFO, on_error=on_error, on_critical=on_critical)

Теперь вызовем событие ошибки. Приведу для этого примитивный код и его вывод:

try:
    a = 1 / 0
except Exception as e:
    mylogger.error(e)

# [2025-09-30 00:00:00 __main__ ERROR exapmle.py:89/<module>] division by zero
# Возникла ошибка

Мы видим, что получили достаточную информацию о нашей ошибке, отформатированную в читаемый вид. Также мы реализовали механизм уведомления о определённых событиях. Это удовлетворяет нашим поставленным желаниям, а следовательно реализация пользовательского обработчика завершено.

Таким образом, мы создали гибкую систему, которая не только улучшает читаемость логов в консоли, но и предоставляет механизм для мгновенного реагирования на критические события.

Исходный код ^[14]