Логирование в Python: настройка уровней, JSON-формата, ротации и сквозного request_id

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

---

Как устроено логирование в Python: база и уровни сообщений

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

• Loggers (Логгеры): Точки входа для записи событий в коде приложения.
• Handlers (Обработчики): Определяют направление вывода (консоль, файл, сокет).
• Filters (Фильтры): Отбирают записи по условиям или обогащают их контекстом.
• Formatters (Форматтеры): Преобразуют внутренний объект LogRecord в итоговую строку.

Иерархия логгеров и стандартные уровни

Логгеры организованы в древовидную иерархию через точечную нотацию (app.services.auth). Корневой логгер называется root. При вызове события передаются вверх по дереву родительским логгерам (механизм propagation), если это явно не отключено свойством propagate = False.

Стандартные уровни логирования имеют следующие числовые значения:

• DEBUG (10) — подробная диагностика для отладки.
• INFO (20) — подтверждение штатной работы ключевых этапов.
• WARNING (30) — предупреждения о непредвиденных ситуациях, не прерывающих работу программы.
• ERROR (40) — сбои в выполнении конкретной операции.
• CRITICAL (50) — критические ошибки, способные привести к аварийной остановке сервиса.

import logging

logger = logging.getLogger(__name__)

logger.debug("Инициализация модуля")
logger.info("Пользователь успешно авторизован")
logger.warning("Время ответа внешнего API превысило 500мс")

Обработка ошибок: почему важно использовать logger.exception()

При перехвате исключений разработчики нередко вызывают logger.error("Ошибка: %s", str(e)). В таком случае теряется трейсбек (стек вызовов), что существенно усложняет локализацию бага.

Метод logger.exception() автоматически записывает событие на уровне ERROR и добавляет полный стек ошибки (exc_info=True):

try:
    result = 10 / 0
except ZeroDivisionError:
    # Запишет стек вызовов автоматически
    logger.exception("Не удалось выполнить деление")

---

Конфигурация через dictConfig: декларативный подход

Настройка логгеров через императивный код быстро становится громоздкой. Функция logging.config.dictConfig позволяет централизованно задать параметры через словарь Python, YAML- или JSON-файл.

Структура словаря: логгеры, обработчики и форматтеры

import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {
            "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "standard",
            "stream": "ext://sys.stdout",
        },
    },
    "loggers": {
        "app": {
            "handlers": ["console"],
            "level": "DEBUG",
            "propagate": False,
        },
    },
    "root": {
        "handlers": ["console"],
        "level": "WARNING",
    },
}

logging.config.dictConfig(LOGGING_CONFIG)

Параметр disable_existing_loggers

Параметр "disable_existing_loggers": False защищает от распространенной проблемы: если сторонние библиотеки (например, urllib3 или SQLAlchemy) создали свои логгеры до вызова dictConfig, значение True отключит их, и приложение перестанет получать важные служебные сообщения.

---

Структурированные JSON-логи

Зачем переходить от текста к JSON

Текстовые строки сложно парсить лог-коллекторам (Fluentbit, Logstash, Vector). Разработчикам приходится писать и поддерживать регулярные выражения (Grok-паттерны), которые ломаются при малейшем изменении формата строки.

JSON-логи изначально сериализуются в пары «ключ-значение». Это позволяет системам хранения (OpenSearch, Elasticsearch, Grafana Loki, ClickHouse) без задержек индексировать отдельные поля: user_id, http_status, duration_ms.

Реализация собственного JSON-форматтера

Создадим форматтер на базе стандартного logging.Formatter и функции json.dumps():

import json
import logging
from datetime import datetime, timezone

class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "line": record.lineno,
        }

        # Извлекаем кастомные данные, переданные через extra
        for key, value in record.__dict__.items():
            if key not in logging.LogRecord.__dict__ and key not in log_data:
                log_data[key] = value

        # Обработка трейсбеков
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_data, ensure_ascii=False)

---

Сквозная трассировка: корреляция запросов по request_id

Проблема поиска логов распределенного запроса

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

Передача контекста через contextvars и logging.Filter

В асинхронных фреймворках (FastAPI, Aiohttp) или многопоточных сервисах глобальные переменные приводят к перезаписи контекста между параллельными корутинами. Для изоляции контекста запроса используется стандартный модуль contextvars.

import logging
from contextvars import ContextVar
import uuid

# Контекстная переменная для хранения ID текущего запроса
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

class RequestIDFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True

Подключение фильтра и использование в коде:

# Настройка логгера
handler = logging.StreamHandler()
handler.addFilter(RequestIDFilter())
handler.setFormatter(logging.Formatter("%(levelname)s [%(request_id)s] %(message)s"))

logger = logging.getLogger("api")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Имитация обработки запроса в middleware
def handle_request():
    token = request_id_var.set(str(uuid.uuid4()))
    try:
        logger.info("Начало обработки запроса")
        # Бизнес-логика...
        logger.info("Запрос успешно обработан")
    finally:
        request_id_var.reset(token)

handle_request()

---

Безопасность логов: маскирование чувствительных данных

Попадание персональных данных (PII), паролей, номеров банковских карт или JWT-токенов в логи создает прямую угрозу безопасности и нарушает требования стандартов (GDPR, PCI-DSS).

С помощью logging.Filter можно перехватывать и санировать LogRecord до этапа сериализации:

import logging
import re

class SensitiveDataFilter(logging.Filter):
    PATTERNS = [
        (re.compile(r'(password["\']?\s*[:=]\s*["\'])([^"\']+)(["\'])', re.IGNORECASE), r'\1***\3'),
        (re.compile(r'(Bearer\s+)[A-Za-z0-9\-._~+/]+=*', re.IGNORECASE), r'\1***'),
    ]

    def filter(self, record: logging.LogRecord) -> bool:
        if isinstance(record.msg, str):
            for pattern, repl in self.PATTERNS:
                record.msg = pattern.sub(repl, record.msg)
        return True

---

Ротация файлов логов

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

Обработчик	Критерий ротации	Основные параметры	Применение
RotatingFileHandler	Размер файла	maxBytes, backupCount	Ограничение максимального объема логов на диске
TimedRotatingFileHandler	Временной интервал	when, interval, backupCount	Архивирование логов по дням, неделям или часам

Ротация по размеру: RotatingFileHandler

Создает новый файл при достижении лимита в байтах (maxBytes) и хранит фиксированное число архивных копий (backupCount):

from logging.handlers import RotatingFileHandler

file_handler = RotatingFileHandler(
    filename="app.log",
    maxBytes=10 * 1024 * 1024,  # 10 МБ
    backupCount=5,              # Файлы app.log, app.log.1 ... app.log.5
    encoding="utf-8",
)

Ротация по времени: TimedRotatingFileHandler

Архивирует логи с заданной периодичностью (например, каждый день в полночь):

from logging.handlers import TimedRotatingFileHandler

timed_handler = TimedRotatingFileHandler(
    filename="app_daily.log",
    when="midnight",     # Ротация каждую полночь
    interval=1,
    backupCount=14,      # Хранить историю за 14 дней
    encoding="utf-8",
)

---

Тестирование логирования с помощью pytest и caplog

Фреймворк pytest содержит встроенную фикстуру caplog, которая перехватывает сообщения логгеров для проверки в автотестах.

import logging
import pytest

def divide(a: int, b: int) -> float:
    logger = logging.getLogger("math_module")
    if b == 0:
        logger.error("Попытка деления на ноль: a=%s", a)
        raise ValueError("Деление на ноль")
    return a / b

def test_divide_zero_logging(caplog):
    # Устанавливаем уровень перехвата
    caplog.set_level(logging.ERROR)

    with pytest.raises(ValueError):
        divide(10, 0)

    # Проверяем вхождение текста в общий лог
    assert "Попытка деления на ноль: a=10" in caplog.text

    # Проверяем точное соответствие кортежей (logger_name, level, message)
    assert ("math_module", logging.ERROR, "Попытка деления на ноль: a=10") in caplog.record_tuples

---

FAQ (Часто задаваемые вопросы)

1. Чем logger.exception() отличается от logger.error()?

Метод logger.exception() вызывает запись с уровнем ERROR и автоматически добавляет exc_info=True. Он извлекает текущий стек вызовов из блока except и прикрепляет его к логу. Если вызвать logger.exception() вне блока except, поле трейсбека будет пустым.

2. Почему в продакшене лучше использовать JSON-логи?

JSON — стандартизированный машинно-читаемый формат. Лог-коллекторы и SIEM-системы парсят его «из коробки» без сложных Grok-шаблонов, что снижает нагрузку на CPU лог-агентов и исключает ошибки при изменении состава полей.

3. Что произойдет, если в форматтере указан %(request_id)s, а в extra он не передан?

Встроенный строковый форматтер Python выбросит исключение KeyError, из-за чего сообщение не будет записано. Для предотвращения таких сбоев используйте logging.Filter, который гарантирует наличие поля со значением по умолчанию (например, -).

4. Как избежать утечки секретов и токенов в файлы логов?

Используйте кастомные logging.Filter для автоматической замены чувствительных паттернов регулярными выражениями или валидируйте входящие DTO через схемы с маскированием (например, SecretStr в Pydantic) перед их логированием.

5. Нужна ли ротация файлов при запуске приложения в Docker?

В контейнеризированных средах (Docker, Kubernetes) принято писать логи в stdout/stderr. Ротацию и сбор логов в таком случае выполняет инфраструктурный слой — драйвер логирования Docker или сборщики Vector/Fluentbit.

---

Заключение

Надежная система логирования в Python опирается на три базовых принципа:

1. Декларативная конфигурация: Используйте dictConfig для централизованного управления обработчиками и уровнями.
2. Структурированный вывод: Сериализуйте логи в JSON для интеграции с внешними хранилищами и анализаторами.
3. Контекст запроса: Внедряйте request_id с помощью contextvars и фильтров, чтобы связать разрозненные события в единый трассировочный след.

Источники

• _pytest.logging - pytest documentation
• Logging — pytest documentation
• Logging¶
• Source code for _pytest.logging - pytest documentation
• docs.pytest.org › en › stableHow to manage logging - pytest documentation
• logging.config — Logging configuration