Представьте ситуацию: ваше приложение работает в продакшене, как вдруг происходит критическая ошибка. Вы узнаете о ней только через несколько часов, когда пользователи начинают массово жаловаться. Идёте проверять консоль, а тут всего лишь трассировка стека, которая мало что говорит о проблеме. Из-за кого и когда возникла это ошибка? Чтобы предотвратить такие сценарии, необходима активная система уведомлений. В этом руководстве мы создадим пользовательский обработчик, который предоставит возможность создания системы уведомлений об ошибках, которая гарантирует, что вы всегда будете в курсе состояния вашей системы.
Теоретическая основа
Модуль logging реализует иерархическую систему логирования, которая состоит из 4-х ключевых компонентов:
Logger — ключевой объект для работы с логами. Имеет имя и уровень логирования (
NOTSET,DEBUG,INFO,WARNING,ERROR,CRITICAL).Handler — определяет, как и куда отправлять сообщения (файл, консоль, сокет, email и т.д.). Один логгер может иметь несколько обработчиков.
Filter — используется для дополнительной фильтрации сообщений (например, пропускать только определённые уровни или сообщения с конкретным текстом).
Formatter — определяет, как будет выглядеть сообщение (дата, уровень, модуль, текст и т.д.).
Сперва при возникновении события формируем запись. Затем мы определяем назначение этой записи, используя её поля. При необходимости фильтруем их. И в завершение форматируем итог в читаемый вид.
Handler — это базовый класс для всех обработчиков логов, в котором некоторые методы определяются только в его потомках. Он отвечает за отправку сообщений из логгера в конечное место назначения.
StreamHandler — это тот самый потомок, предназначенный для логирования ошибок в заданные потоки, некоторая часть кода из него нам понадобиться.
LogRecord — структура данных, которая содержит всю информацию о конкретном событии логирования.
Основные методы класса Handler:
format — Преобразует объект LogRecord в строку и возвращает её в отформатированном виде.
emit — Определяет, как именно сообщение будет выведено (например, вывод в консоль, запись в файл).
handle — Проверяет фильтры и уровень записи, вызывает emit, если запись прошла проверки.
Чего мы хотим?
Для начала нам следует определить, что мы хотим от нашего пользовательского обработчика. Выделю несколько пунктов:
Получать достаточную информацию в консоли, отформатированная в читаемый вид, для определения и устранения проблем и отладки приложения (например, время, модуль, уровень и текст ошибки)
Иметь возможность получать уведомления в зависимости от уровня события (например, стандартные ошибки записывать в специальный файл, а при критических ошибках отправлять уведомление разработчику)
Пишем код
Начнём с определения нового класса и его конструктора. Он будет унаследован от Handler. Для реализации 2-го пункта нам нужны callback-функции, которые будут вызываться при обработке события определённого уровня. Сохраним их в словаре, где ключом будет уровень логирования.
LogCallback = Optional[Callable[[LogRecord], None]] class LoggingHandler(Handler): def __init__( self, on_critical: LogCallback = None, on_error: LogCallback = None, on_warning: LogCallback = None, on_info: LogCallback = None, on_debug: LogCallback = 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)
Реализацию констант, приведённых выше, я вам предлагаю реализовать самостоятельно. Атрибуты для форматирования текста и временных меток вам в помощь. Ниже я приведу пример из своего приложения:
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, }
Приступим теперь к логике нашего обработчика. Для этого нам следует переопределить метод handle. Делать он будет тоже, что и метод в супперклассе, однако в конце мы добавим вызов наших callback-функций.
def handle(self, record: LogRecord): rv = super().handle(record) if not rv: return rv callback = self.callbacks.get(record.levelno) if callback: callback(record) return rv
Дополнительно я хочу, чтобы полоса загрузки в tqdm корректно отображалась в логах. Для этого копируем реализацию метода emit из класса StreamHandler и модифицируем его под наши нужды.
def emit(self, record: LogRecord): try: msg = self.format(record) tqdm.write(msg, end=self.terminator) except RecursionError: raise except Exception: self.handleError(record)
Теперь нам нужно написать метод для его установки. В аргументы пойдёт всё нужное для класса Logger и методы, которые будут вызваны при определённом уровне события.
def setup_logger( name: str, level: int = logging.INFO, on_critical: LogCallback = None, on_error: LogCallback = None, on_warning: LogCallback = None, on_info: LogCallback = None, on_debug: LogCallback = None, ) -> logging.Logger: logger = logging.getLogger(name) logger.setLevel(level) if not logger.handlers: handler = LoggingHandler( 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 # Возникла ошибка
Мы видим, что получили достаточную информацию о нашей ошибке, отформатированную в читаемый вид. Также мы реализовали механизм уведомления о определённых событиях. Это удовлетворяет нашим поставленным желаниям, а следовательно реализация пользовательского обработчика завершена.
Таким образом, мы создали гибкую систему, которая не только улучшает читаемость логов в консоли, но и предоставляет механизм для мгновенного реагирования на критические события.
