Представьте ситуацию: ваше приложение работает в продакшене, как вдруг происходит критическая ошибка. Вы узнаете о ней только через несколько часов, когда пользователи начинают массово жаловаться. Идёте проверять консоль, а тут всего лишь трассировка стека, которая мало что говорит о проблеме. Из-за кого и когда возникла это ошибка? Чтобы предотвратить такие сценарии, необходима активная система уведомлений. В этом руководстве мы создадим пользовательский обработчик, который предоставит возможность создания системы уведомлений об ошибках, которая гарантирует, что вы всегда будете в курсе состояния вашей системы.
Теоретическая основа
Модуль logging
реализует иерархическую систему логирования, которая состоит из 4-х ключевых компонентов:
Logger
— ключевой объект для работы с логами. Имеет имя и уровень логирования (DEBUG
,INFO
,WARNING
,ERROR
,CRITICAL
).Handler
— определяет, как и куда отправлять сообщения (файл, консоль, сокет, email и т.д.). Один логгер может иметь несколько обработчиков.Filter
— используется для дополнительной фильтрации сообщений (например, пропускать только определённые уровни или сообщения с конкретным текстом).Formatter
— определяет, как будет выглядеть сообщение (дата, уровень, модуль, текст и т.д.).
Handler
— это базовый класс для всех обработчиков логов, в котором некоторые методы определяются только в его потомках. Он отвечает за отправку сообщений из логгера в конечное место назначения.StreamHandler
— это тот самый потомок, предназначенный для логирования ошибок в консоль, его мы и будем его переопределять.LogRecord
— структура данных, которая содержит всю информацию о конкретном событии логирования.
Основные методы класса Handler
:
format
— Преобразует объектLogRecord
в строку и возвращает её в отформатированном виде.emit
— Определяет, как именно сообщение будет выведено (например, вывод в консоль, запись в файл).handle
— Проверяет фильтры и уровень записи, вызываетemit
, если запись прошла проверки.
Чего мы хотим?
Для начала нам следует определить, что мы хотим от нашего пользовательского обработчика. Выделю несколько пунктов:
Получать достаточную информацию в консоли, отформатированная в читаемый вид, для определения и устранения проблем и отладки приложения (например, время, модуль, уровень и текст ошибки)
Иметь возможность получать уведомления в зависимости от уровня события (например, стандартные ошибки записывать в специальный файл, а при критических ошибках отправлять уведомление разработчику)
Пишем код
Начнём с определения нового класса и его конструктора. Он будет унаследован от 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)
Реализацию констант, приведённых выше, я вам предлагаю реализовать самостоятельно. Атрибуты для форматирования текста и временных меток вам в помощь. Ниже я приведу пример из своего приложения:
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
, чтобы полоса загрузки в логах отображалась снизу. Вы же можете использовать любой из доступных способов вывода информации в консоль. Так же тут мы и вызываем нужный нам метод при обработке события.
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
# Возникла ошибка
Мы видим, что получили достаточную информацию о нашей ошибке, отформатированную в читаемый вид. Также мы реализовали механизм уведомления о определённых событиях. Это удовлетворяет нашим поставленным желаниям, а следовательно реализация пользовательского обработчика завершена.
Таким образом, мы создали гибкую систему, которая не только улучшает читаемость логов в консоли, но и предоставляет механизм для мгновенного реагирования на критические события.