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

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

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

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

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

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

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

Сперва при возникновении события формируем запись. Затем мы определяем назначение этой записи, используя её поля. При необходимости фильтруем их. И в завершение форматируем итог в читаемый вид.

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

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

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

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

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

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

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

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

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

Пишем код

Начнём с определения нового класса и его конструктора. Он будет унаследован от 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
# Возникла ошибка

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

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

Исходный код