izhatomic 9 апр в 16:15

Тонкости работы с логгированием в Python: краткий гайд для разработчиков

Простой

13 мин

7.9K

Python*Анализ и проектирование систем*ООП*Программирование*Веб-разработка*

Логирование является одним из ключевых и важнейших элементов разработки и эксплуатации приложений. Оно дает ценную информацию всей цепочке заинтересованных лиц: от разработчиков и системных администраторов (отладка, мониторинг и т.д.) до руководителей бизнеса в целом (поведение пользователей, соблюдение нормативных требований и т.д.).

Давайте разберем, как работает де-факто стандарт логирования — модуль logging в Python. Несмотря на то что это довольно мощный инструмент, он сложный и в чем-то немного устаревший: не всегда соответствует python style и не все его фичи вообще следует использовать. Ниже я дам общие рекомендации и опишу практики применения модуля, подходящие для большинства случаев.

Часто можно встретить использование модуля без попытки как-либо его настроить (здесь и далее в статье все примеры кода для Python 3.12):

Пример использования модуля logging

import loggigng
logger = logging.getLogger("app")

def my_function(a):
	logger.info(f"var a set to {a}")
	try:
    	b = int(a)
	except ValueError:
    	logger.error("var a is not integer")
	...

Однако есть более удобный способ — использование конфигурации в виде словаря.

Конфигурация logging в виде словаря

{
  "version": 1,
  "disable_existing_loggers": false,
  "formatters": {
	"simple": {
  	"format": "%(asctime)s - %(levelname)s : %(message)s"
	},
	"detailed": {
  	"format": "%(asctime)s - %(levelname)s - %(module)s(L%(lineno)d) : %(message)s",
  	"datefmt": "%Y-%m-%dT%H:%M:%S%z"
	}
  },
  "handlers": {
	"stderr": {
  	"class": "logging.StreamHandler",
  	"level": "WARNING",
  	"formatter": "simple"
	},
	"file": {
  	"class": "logging.handlers.RotatingFileHandler",
  	"level": "DEBUG",
  	"formatter": "detailed",
  	"filename": "logs/my_app.log",
  	"maxBytes": 1000000,
  	"backupCount": 3
	}
  },
  "loggers": {
	"root": {
  	"level": "DEBUG",
  	"handlers": [
    	"stderr",
    	"file"
  	]
	}
  }
}

Настройка таким способом позволяет в явном и наглядном виде параметризировать все компоненты ведения журнала: фильтры (filter), форматеры (formatter), обработчики (handler) и регистраторы (logger). У всех этих компонентов есть настройки «по-умолчанию», которые как раз и позволяют в две строчки имплементировать модуль в свой код и использовать его без конфигурирования. Однако для более эффективного ведения журнала всё это можно настроить под себя и свои потребности. Чтобы понимать, что мы будем настраивать дальше и как эти компоненты взаимодействуют между собой, коротко разберем их предназначение.

Фильтры (filters)

Позволяют регулировать обработку сообщений. Например, можно отбрасывать сообщения, содержащие определенные слова. Также можно настроить более сложное поведение: например, отбирать сообщения по определенному критерию (содержат чувствительные данные) и изменять их содержимое на лету.

Обработчики (handlers)

Отвечают за то, как и где регистрировать сообщение. Это может быть вывод в STDOUT, отправка сообщения по сети, используя syslog, запись в файл и т.д. Так же, как и для регистратора, для обработчика можно настроить уровень сообщений и фильтры. Если сообщение было отброшено одним обработчиком, оно все равно будет передано другим обработчикам. Таким образом, можно настроить, какие сообщения и куда сохранять. Полностью отбросить сообщение можно на уровне регистратора. Чаще всего сообщение отправляется сразу в несколько мест, как раз для этого и создается несколько обработчиков.

Форматер (formatters)

Чаще всего запись в журнале — это какое-то простое текстовое сообщение. Однако в любом случае переданное в коде сообщение преобразуется в объект LogRecord, который имеет ряд полезных атрибутов, позволяющих гибко настроить внесение записи в журнал: указать уровень, временную метку, место в коде, откуда эта запись была сделана и т.д. В результате запись может быть внесена в журнал в виде строки (STDOUT) или в виде JSON-объекта. Именно форматеры содержат бо́льшую часть настроек, отвечающих за то, в каком виде сообщение попадет в журнал.

Регистраторы (loggers)

Фактически именно эту сущность мы и используем непосредственно в коде. У регистратора есть набор методов (.info() , .warning() , .debug() etc), которые позволяют записать в журнал различные типы сообщений. Каждый регистратор в свою очередь можно гибко настраивать: указывать уровень логирования, отбрасывая ненужные сообщения, использовать фильтры сообщений и т.д.

Регистраторов может быть несколько. Обязательно есть главный корневой регистратор, а дальше от него в виде древовидной структуры могут отделяться и существовать другие регистраторы. Каждый регистратор имеет свое уникальное имя. Когда сообщение попадает в определенный регистратор, оно проходит через все его обработчики и передается дальше родительскому регистратору. Таким образом, любое сообщение, переданное в дочерний регистратор, дойдет до главного корневого регистратора (если передача «наверх» не запрещена в явном виде в настройках).

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

Рассмотрим схему прохождения сообщения через все элементы системы логирования.

Как видно из схемы, сообщение, переданное в регистратор, проходит через проверку уровня и фильтрацию и далее передается во все обработчики. Даже если оно будет заблокировано (по уровню или в фильтрах) в одном (или во всех!) из обработчиков, оно все равно будет передано дальше в родительский регистратор и так далее, пока не достигнет корневого регистратора.

Создание большого количества регистраторов, каждый со своим набором обработчиков и фильтров, зачастую является избыточным. Чаще всего наиболее оптимальным решением является структура ниже:

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

Не стоит использовать корневой регистратор для записей в журнал. Создайте свой регистратор (дочерний от корневого), используя метод logging.getlogger("my_first_registrator") и передав ему имя регистратора. Все сообщения, переданные в ваш регистратор, будут попадать в корневой и обрабатываться там, поэтому не создавайте никаких обработчиков в вашем регистраторе, просто настройте корневой. Если приложение очень большое, то создайте еще несколько не-корневых регистраторов и используйте их в компонентах своего приложения. Не стоит делать отдельные регистраторы для каждого файла своего приложения, это тоже избыточно. Просто сделайте свой регистратор глобальным и используйте в нескольких местах.

Создать структуру регистраторов из рекомендуемой схемы можно с помощью кода:

logger1 = logging.getLogger("COMPONENT_1")
logger2 = logging.getLogger("COMPONENT_2")

Создать дочерний регистратор от COMPONENT_1 можно так:

child_logger1 = logger1.getChild("child")

или так:

child_logger1 = logging.getLogger("COMPONENT_1.child")

Теперь перейдем от теории к практике. Вот так будет выглядеть словарь с настройками для простейшей конфигурации журналирования с выводом сообщений в STDOUT и простым форматером.

Скрытый текст

{
  "version": 1,
  "disable_existing_loggers": false,
  "formatters": {
	"simple": {
  	"format": "%(levelname)s: %(message)s"
	}
  },
  "handlers": {
	"stdout": {
  	"class": "logging.StreamHandler",
  	"formatter": "simple"
	}
  },
  "loggers": {
	"root": {
  	"level": "DEBUG",
  	"handlers": [
    	"stdout"
  	]
	}
  }
}

Форматер определяет внешний вид строки. Мы не указываем здесь никакого специализированного форматера и используем стандартный, в котором есть набор предопределенных переменных, с помощью которых мы можем определить внешний вид записи. Список стандартных переменных можно найти в документации. Также стоит отметить, что словарь с конфигурацией может храниться и в формате YAML, однако мне кажется более удобным использовать JSON, хотя бы просто потому, что не нужно импортировать дополнительные модули и использовать только стандартные инструменты Python. Также, по моему мнению, в YAML проще ошибиться. Впрочем, формат конфигурации всегда остается на ваше усмотрение.

В случае, когда приложение достаточно сложное и простых строковых сообщений недостаточно, можно хранить сообщения в формате JSON. Это позволит вам обогатить их большим количеством данных, структурировать их содержимое и легко анализировать записи. JSON в данном случае — это обычный формат записи, а значит, нам достаточно добавить сюда соответствующий форматер. Существует ряд готовых решений, которые вы можете установить с помощью pip, однако для демонстрационных целей сделаем свой простой JSON-форматер.

Пример JSON-форматера

import logging
import datetime as dt
import json

class MyJSONFormatter(logging.Formatter):
	DEFAULT_TIMEZONE = dt.timezone.utc

	def __init__(self, *, format_keys: dict[str, str] | None = None):
    	super().__init__()
    	self.format_keys = format_keys if format_keys is not None else {}

	def format(self, record: logging.LogRecord) -> str:
    	formatted_message = self._build_log_message(record)
    	return json.dumps(formatted_message, default=str)

	def _build_log_message(self, record: logging.LogRecord):
    	core_fields = {
        	"message": record.getMessage(),
        	"timestamp": self._format_timestamp(record.created),
    	}
    	self._format_optional_fields(record, core_fields)

    	formatted_message = {
        	key: core_fields.pop(value, getattr(record, value, None))
        	for key, value in self.format_keys.items()
    	}
    	formatted_message.update(core_fields)

    	for key, value in record.__dict__.items():
        	if key not in LOG_RECORD_BUILTIN_ATTRS:
            	formatted_message[key] = value

    	return formatted_message

	def _format_timestamp(self, timestamp: float) -> str:
    	return dt.datetime.fromtimestamp(timestamp, tz=self.DEFAULT_TIMEZONE).isoformat()

	def _format_optional_fields(self, record: logging.LogRecord, fields: dict):
    	if record.exc_info is not None:
        	fields["exc_info"] = self.formatException(record.exc_info)
    	if record.stack_info is not None:
        	fields["stack_info"] = self.formatStack(record.stack_info)

Важное уточнение: при записи таких сообщений в файл, сам он не будет являться корректным JSON-файлом. Корректными JSON-записями являются строки этого файла, поэтому лучше хранить такие данные в файле с расширением .jsonl и при дальнейшем разборе читать содержимое построчно и анализировать каждую строку как JSON.

Также обратите внимание, что в запись в формате JSON становится возможным добавлять много дополнительной контекстной информации. Для этого мы можем просто использовать дополнительный аргумент в вызове журнала, присвоить ему словарь с дополнительной информацией, и эта запись останется в журнале. Пример:

logger.debug("debug message", extra={"MyKey": "MyVar"})

В журнале появится запись наподобие этой:

{"level": "DEBUG", "message": "debug message", "timestamp": "2025-04-03T17:44:09.464604+00:00", "logger": "my_app", "module": "test", "function": "main", "line": 104, "thread_name": "MainThread", "MyKey": "MyVar"}

Фильтры создаются аналогичным образом. Пример фильтра, который будет отбрасывать сообщения, содержащие в себе подстроку "secret":

Пример простого фильтра

class WordFilter(logging.Filter):
	@override
	def filter(self, record: logging.LogRecord) -> bool | logging.LogRecord:
    	message = record.getMessage()
    	if "secret" in message.lower():
        	return False

    	return True

Как видно из примера, для создания фильтра нам необходимо отнаследовать свой класс от logging.Filter и переопределить метод filter, принимающий экземпляр записи logging.LogRecord . Фильтр либо может вернуть булево значение, тогда False не пропустит сообщение дальше, соответственно True пропустит, либо может вернуть экземпляр записи в лог, тогда далее в обработчик попадет уже этот экземпляр. На основе этого реализуются фильтры, способные изменять содержимое. Например:

Пример фильтра, изменяющего содержимое записи

class TransformFilter(logging.Filter):
	@override
	def filter(self, record: logging.LogRecord) -> bool | logging.LogRecord:
    	message = record.getMessage()
    	if "telephone" in message.lower():
        	new_rec = deepcopy(record)
        	new_rec.msg = message.replace("telephone", "phone")

        	return new_rec

    	return True

В этом случае в журнал попадет сообщение, где telephone заменен на phone. Конечно же, внутри фильтра вы можете реализовать и более сложную логику. Обратите внимание, что у меня используется создание deepcopy исходной записи. Если изменить содержимое исходной записи (record), то далее во все обработчики пойдет уже видоизмененная запись. Если это ожидаемое поведение, копировать первоначальное сообщение не нужно. Но если вы хотите изменить запись только для тех обработчиков, где применяется этот фильтр, то копию сделать придется, чтобы не сломать работу журналирования в других местах.

Несколько слов о производительности. Каждая операция записи в журнал — это операция ввода-вывода. Не важно, пишите вы в локальный файл, отправляете сообщение по сети с помощью rsyslog — это IO-операция, и она может занимать существенное время. Если у вас какой-нибудь веб-сервер, который отвечает на внешние запросы, то каждая запись в лог будет увеличивать время ответа сервера и снижать его производительность. Чтобы сделать операции записи в журнал не блокирующими, мы можем использовать специальный обработчик — QueueHandler, точнее logging.handlers.QueueHandler. Он сохраняет передаваемые ему записи в очередь, а далее передает их в целевые обработчики в отдельном потоке, таким образом записывая их в «фоновом режиме», не блокируя основную работу приложения. Пример конфигурации с использованием QueueHandler:

Конфигурация с QueueHandler

{
  "version": 1,
  "disable_existing_loggers": false,
  "formatters": {
	"json": {
  	"()": "mylogger.MyJSONFormatter",
  	"fmt_keys": {
    	"level": "levelname",
    	"message": "message",
    	"timestamp": "timestamp",
    	"logger": "name",
    	"module": "module",
    	"function": "funcName",
    	"line": "lineno",
    	"thread_name": "threadName"
  	}
	}
  },
  "handlers": {
	"stderr": {
  	"class": "logging.StreamHandler",
  	"level": "WARNING",
  	"formatter": "json"
	},
	"queue_handler": {
  	"class": "logging.handlers.QueueHandler",
  	"handlers": [
    	"stderr"
  	],
  	"respect_handler_level": true
	}
  },
  "loggers": {
	"root": {
  	"level": "DEBUG",
  	"handlers": [
    	"queue_handler"
  	]
	}
  }
}

Обратите внимание на параметр "respect_handler_level": по умолчанию его значение равно False. Этот параметр отвечает за проверку уровня логирования до передачи сообщения в целевые обработчики. Сообщение будет передано в очередь только в том случае, если его уровень соответствует уровню целевых обработчиков. Например, если пришло сообщение с уровнем DEBUG, а у двух ваших обработчиков настроены уровни INFO и ERROR, то сообщение будет отброшено еще на этапе записи в очередь. Установка параметра в значение True позволит минимизировать объем передаваемых через очередь данных и избегать обработки сообщений, которые всё равно будут отброшены целевыми обработчиками. При больших объемах данных это может значительно снизить нагрузку.

Еще один важный момент, касающийся использования QueueHandler. Поскольку его выполнение происходит в отдельном потоке, этот поток необходимо запустить вручную самостоятельно, сам он автоматически не запускается. Также нам надо будет корректно завершить его при завершении работы приложения. Сделать это можно, получив обработчик по имени:

Пример запуска QueueHandler

import atexit
import logging

queue_handler = logging.getHandlerByName("queue_handler")
	if queue_handler is not None:
    	queue_handler.listener.start()
    	atexit.register(queue_handler.listener.stop)

При желании можно сделать свой класс-обертку (отнаследовав его от logging.handlers.QueueHandler) и реализовать внутри него всю логику работы с запуском и остановкой потоков. В этом случае вся работа будет спрятана внутрь и полностью автоматизирована.

Итоговый код с примерами:

main.py

import logging
import logging.config
import atexit

from LoggerAddons import NonErrorFilter, FilterSec
import logging.handlers


config = {
  "version": 1,
  "disable_existing_loggers": False,
  "formatters": {
    "simple": {
      "format": "%(asctime)s - %(levelname)s - %(module)s(L%(lineno)d) : %(message)s",
      "datefmt": "%Y-%m-%dT%H:%M:%S%z"
    },
    "json": {
      "()": "mylogger.MyJSONFormatter",
      "format_keys": {
        "level": "levelname",
        "message": "message",
        "timestamp": "timestamp",
        "logger": "name",
        "module": "module",
        "function": "funcName",
        "line": "lineno",
        "thread_name": "threadName"
      }
    }
  },
  "handlers": {
    "stderr": {
      "class": "logging.StreamHandler",
      "level": "DEBUG",
      "formatter": "simple",
      "filters": [NonErrorFilter(), FilterSec()]
    },
    "file_json": {
      "class": "logging.handlers.RotatingFileHandler",
      "level": "DEBUG",
      "formatter": "json",
      "filename": "my_app.log.jsonl",
      "maxBytes": 10000,
      "backupCount": 3,
      "filters": [NonErrorFilter()]
    },
    "queue_handler": {
      "class": "logging.handlers.QueueHandler",
      "handlers": [
        "stderr",
        "file_json"
      ],
      "respect_handler_level": True
    }
  },
  "loggers": {
    "root": {
      "level": "DEBUG",
      "handlers": [
        "queue_handler"
      ]
    }
  }
}


def setup_logging(logger_name: str) -> logging.Logger:
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.DEBUG)

    logging.config.dictConfig(config)

    queue_handler = logging.getHandlerByName("queue_handler")
    if queue_handler is not None:
        queue_handler.listener.start()
        atexit.register(queue_handler.listener.stop)

    return logger


def main():
    logger = setup_logging("my_app")

    logger.debug("debug message", extra={"MyKey": "MyVar"})
    logger.info("info message: telephone +7962909")
    logger.warning("warning message")
    logger.error("error message")
    logger.critical("critical message")

    try:
        1 / 0
    except ZeroDivisionError:
        logger.exception("exception message")

    child_logger = logger.getChild("child")
    # child_logger = logging.getLogger("my_app.child")

    child_logger.addFilter(FilterSec())
    child_logger.error("## --> child error message")
    child_logger.debug("## --> child debug message")
    child_logger.info("## --> child info message with telephone +7962909")


if __name__ == "__main__":
    main()

LoggerAddons.py

import datetime as dt
import json
from copy import deepcopy
import logging.handlers
from typing import override


LOG_RECORD_BUILTIN_ATTRS = {
    "args",
    "asctime",
    "created",
    "exc_info",
    "exc_text",
    "filename",
    "funcName",
    "levelname",
    "levelno",
    "lineno",
    "module",
    "msecs",
    "message",
    "msg",
    "name",
    "pathname",
    "process",
    "processName",
    "relativeCreated",
    "stack_info",
    "thread",
    "threadName",
    "taskName",
}


class MyJSONFormatter(logging.Formatter):
    DEFAULT_TIMEZONE = dt.timezone.utc

    def __init__(self, *, format_keys: dict[str, str] | None = None):
        super().__init__()
        self.format_keys = format_keys if format_keys is not None else {}

    def format(self, record: logging.LogRecord) -> str:
        formatted_message = self._build_log_message(record)
        return json.dumps(formatted_message, default=str)

    def _build_log_message(self, record: logging.LogRecord):
        core_fields = {
            "message": record.getMessage(),
            "timestamp": self._format_timestamp(record.created),
        }
        self._format_optional_fields(record, core_fields)

        formatted_message = {
            key: core_fields.pop(value, getattr(record, value, None))
            for key, value in self.format_keys.items()
        }
        formatted_message.update(core_fields)

        for key, value in record.__dict__.items():
            if key not in LOG_RECORD_BUILTIN_ATTRS:
                formatted_message[key] = value

        return formatted_message

    def _format_timestamp(self, timestamp: float) -> str:
        return dt.datetime.fromtimestamp(timestamp, tz=self.DEFAULT_TIMEZONE).isoformat()

    def _format_optional_fields(self, record: logging.LogRecord, fields: dict):
        if record.exc_info is not None:
            fields["exc_info"] = self.formatException(record.exc_info)
        if record.stack_info is not None:
            fields["stack_info"] = self.formatStack(record.stack_info)


class NonErrorFilter(logging.Filter):
    @override
    def filter(self, record: logging.LogRecord) -> bool | logging.LogRecord:
        message = record.getMessage()
        if "error" in message.lower():
            return False

        return True


class FilterSec(logging.Filter):
    @override
    def filter(self, record: logging.LogRecord) -> bool | logging.LogRecord:
        message = record.getMessage()
        if "telephone" in message.lower():
            new_rec = deepcopy(record)
            new_rec.msg = message.replace("+7962", "7XXX")

            return new_rec

        return True

В качестве послесловия. Выше изложены обобщенные рекомендации по работе с журналированием. Не стоит слепо следовать этим советам, особенно если вы точно знаете, что вам нужно. В каждом конкретном случае вы можете и будете отступать от этих общих правил. Никто, кроме вас, не знает, что лучше в вашем конкретном случае и для вашего кода. Если вы уверены в своих действиях и понимаете, как сделать эффективнее, делайте это! И обязательно поделитесь с комьюнити.

Хабы:

Тонкости работы с логгированием в Python: краткий гайд для разработчиков

Публикации

Работа

Ближайшие события