Логирование в Python: structlog, уровни, contextvars и защита персональных данных

← назад к разделу

Логи — это первое, куда смотришь, когда что-то пошло не так. Если они написаны через print, не содержат trace_id и не структурированы — расследование превращается в часы grep-а. В этой статье разберём, как правильно настроить логирование в Python-сервисе: библиотека, формат, уровни, контекст запроса и защита чувствительных данных.

Почему print и logging.getLogger не годятся для продакшна

Начинающий питонист обычно логирует так:

print(f"Order confirmed: {order.id}")

Проблема: в продакшне этого не видно в системе сбора логов (Loki, ELK, Datadog) — они ожидают JSON с полями, а не произвольный текст. Даже logging.getLogger из стандартной библиотеки по умолчанию пишет текст без структуры.

Стандартная библиотека logging можно настроить на JSON, но это сложно и многословно. Вместо этого в Python-сервисах используют structlog — библиотеку, которая с самого начала заточена на структурированные логи.

import structlog

log = structlog.get_logger(__name__)

log.info("order_confirmed", order_id=order.id, customer_id=order.customer_id)

Вместо строки "Order confirmed: ord-9182" в JSON появятся отдельные поля: order_id, customer_id, event, timestamp, level. Эти поля можно фильтровать, агрегировать и искать в любой системе наблюдаемости.

Два профиля: JSON в продакшне, текст при разработке

В продакшне логи должны быть в JSON — одна строка на запись, все поля отдельными ключами. Loki, ELK и Datadog парсят их напрямую без регулярных выражений.

При локальной разработке JSON неудобен для чтения. structlog умеет переключаться между форматами через переменную окружения:

import logging
import structlog
import os

def configure_logging() -> None:
    shared_processors: list[structlog.types.Processor] = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
    ]

    if os.getenv("APP_ENV", "dev") == "prod":
        renderer = structlog.processors.JSONRenderer()
    else:
        renderer = structlog.dev.ConsoleRenderer()

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    formatter = structlog.stdlib.ProcessorFormatter(
        processors=[
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            renderer,
        ],
        foreign_pre_chain=shared_processors,
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.handlers = [handler]
    root_logger.setLevel(logging.INFO)

configure_logging() вызывается один раз при старте приложения — например, в main.py до создания FastAPI-приложения.

При APP_ENV=dev — цветной ConsoleRenderer с выравниванием, удобный для чтения. При APP_ENV=prod — JSONRenderer одной строкой.

Как получать логгер

Одна строка на модуль, в начале файла:

import structlog

log = structlog.get_logger(__name__)


class OrderService:

    def confirm(self, order_id: str) -> Order:
        log.info("confirming_order", order_id=order_id)
        order = self._repository.get(order_id)
        order.confirm()
        self._repository.save(order)
        return order

__name__ автоматически даёт поле logger в JSON со значением вроде order.service — путь к модулю. Это позволяет фильтровать логи по компоненту без дополнительных настроек.

kwargs вместо f-string: почему это важно

Частая ошибка — передавать форматированную строку вместо именованных аргументов:

# Плохо — форматирование выполняется всегда, даже если уровень DEBUG выключен
log.debug(f"Order full state: {order.to_dict()}")

# Хорошо — structlog передаёт объект, рендеринг только при активном уровне
log.debug("order_full_state", order_data=order)

Разница критична для уровня DEBUG с тяжёлой сериализацией: f-string вычисляется при каждом вызове, а kwargs — только если этот уровень включён. На нагруженном сервисе это тысячи лишних сериализаций в секунду.

Кроме производительности, kwargs дают отдельные поля в JSON, по которым можно искать. F-string прячет данные внутри текста — их не отфильтруешь по order_id.

Уровни логов: что куда писать

Частая ошибка — писать INFO на каждый входящий HTTP-запрос в обработчике. В итоге 80% логов — бесполезный шум без контекста. Каждый уровень имеет чёткую семантику:

Уровень	Когда использовать
`ERROR`	Необработанное исключение, недоступность внешнего ресурса с потерей данных. Всегда с `exc_info=True`.
`WARNING`	Деградация с восстановлением: Circuit Breaker сработал, попытка повтора, задействован запасной путь.
`INFO`	Важное бизнес-событие: «order_confirmed», «customer_registered», старт/останов приложения.
`DEBUG`	Детали для отладки. В продакшне выключено, включается по конкретному логгеру при разборе инцидента.

log.error("payment_charge_failed", payment_id=payment_id, exc_info=True)
log.warning("circuit_breaker_open", provider="sber-acquiring", fallback="queue")
log.info("order_confirmed", order_id=order.id, customer_id=order.customer_id, amount=str(order.amount))
log.debug("order_aggregate_state", order_id=order.id, state=order.status)

HTTP-запросы логирует отдельный middleware через uvicorn.access, не обработчики эндпоинтов.

contextvars: сквозные поля без ручной передачи

В Java для сквозных полей (trace_id, request_id, user_id) есть MDC. В Python его роль выполняет structlog.contextvars поверх стандартных contextvars из asyncio.

Идея простая: в начале запроса привязываем поля к контексту потока (или asyncio-задачи), и они автоматически появляются в каждой лог-записи внутри этого запроса — без ручной передачи через параметры.

import uuid
from structlog.contextvars import bind_contextvars, clear_contextvars
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request


class RequestContextMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request: Request, call_next):
        clear_contextvars()
        request_id = request.headers.get("X-Request-Id") or str(uuid.uuid4())
        bind_contextvars(request_id=request_id)
        try:
            return await call_next(request)
        finally:
            clear_contextvars()

clear_contextvars() в finally обязателен: без него поля от предыдущего запроса могут «протечь» в следующий при переиспользовании воркеров.

После JWT-валидации добавляем user_id к контексту:

from structlog.contextvars import bind_contextvars


async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    user = verify_token(token)
    bind_contextvars(user_id=user.id)
    return user

Итоговая запись в продакшне содержит все поля автоматически:

{
  "timestamp": "2026-06-18T14:22:05.123Z",
  "level": "info",
  "logger": "order.service",
  "event": "order_confirmed",
  "order_id": "ord-9182",
  "customer_id": "cust-42",
  "trace_id": "5e92c8a3b1f4d2e6a7c8e9f0a1b2c3d4",
  "span_id": "1f2e3d4c5b6a7980",
  "request_id": "0193a8f3-7c21-7e3f-9b4a-81a2d5f9c030",
  "user_id": "user-42"
}

trace_id и span_id появляются при добавлении OTel-structlog-процессора в цепочку structlog.configure. merge_contextvars сам по себе добавляет только request_id и user_id.

Где ставить логи

Логировать нужно на границах: там, где сервис взаимодействует с внешним миром.

Входящие запросы — access-log через middleware (uvicorn.access). INFO только для критичных команд (платежи, финансовые операции).
Исходящие HTTP-запросы — INFO на запрос, WARNING на ответы 4xx/5xx, ERROR при сетевой ошибке.
Доменные события — INFO при публикации: log.info("event_published", event="OrderCreated", order_id=...).
Фоновые задачи — INFO на старт и завершение с итогом: log.info("outbox_relay_done", published=100, duration_ms=50).

class SberAcquiringAdapter:

    async def charge(self, order_id: str, amount: Decimal) -> ChargeResult:
        log.info("charging_payment", order_id=order_id, amount=str(amount))
        try:
            response = await self._client.post(
                "/charge",
                json={"order_id": order_id, "amount": str(amount)},
            )
            response.raise_for_status()
            log.info("payment_charged", order_id=order_id, status="success")
            return ChargeResult.from_response(response.json())
        except httpx.HTTPStatusError as exc:
            log.warning(
                "payment_charge_http_error",
                order_id=order_id,
                status_code=exc.response.status_code,
            )
            raise
        except httpx.RequestError:
            log.error("payment_charge_network_failure", order_id=order_id, exc_info=True)
            raise

Внутри бизнес-логики — логи только при принятии важного решения или деградации. Не «entering method», не «loaded N rows».

Защита персональных данных в логах

Email, телефон, паспортные данные, токены — всё это нельзя писать в логи. Логи доступны всей команде, хранятся месяцами и индексируются во внешних системах. Попадание персональных данных в логи — это утечка.

# Плохо — персональные данные попадут в Loki/ELK, доступны всей команде
log.info("customer_registered", email=customer.email, phone=customer.phone)

# Хорошо — только внутренний идентификатор
log.info("customer_registered", customer_id=customer.id)

# Если нужна диагностика — маскировать
log.info(
    "email_verification_sent",
    customer_id=customer.id,
    email_mask=mask_email(customer.email),  # c***@sber.ru
)

То же касается тела запроса на финансовых эндпоинтах: не логировать полный request.body(), только идентификаторы (order_id, amount).

Исключения: не теряйте стек

Частая ошибка при логировании ошибок — писать только сообщение, без стека вызовов:

# Плохо — стек потерян, в JSON только event-строка
log.error("product_fetch_failed", product_id=product_id, error=str(exc))

# Хорошо — стек в JSON как stack_trace
log.error("product_fetch_failed", product_id=product_id, exc_info=True)

# Эквивалент внутри except-блока
except Exception:
    log.exception("product_fetch_failed", product_id=product_id)

log.exception(...) — это log.error(...) с exc_info=True по умолчанию. Используйте любой из вариантов, главное — не оставлять ERROR без трассировки.

Коротко

Используйте structlog.get_logger(__name__) вместо print и logging.getLogger.
В продакшне — JSONRenderer, при разработке — ConsoleRenderer; переключается через APP_ENV.
Структурные поля передавайте как kwargs (order_id=order.id), не f-string — это и производительнее, и фильтруемо в поиске.
ERROR — только с exc_info=True; INFO — важные бизнес-события, не каждый HTTP-запрос.
Контекст запроса (request_id, user_id, trace_id) привязывайте через bind_contextvars в middleware, не в обработчиках.
Не забывайте clear_contextvars() в finally — иначе поля протекут между запросами.
Персональные данные (email, телефон, токены) в логах запрещены; логируйте только идентификаторы или маскированные значения.

Что почитать дальше

Tracing — OpenTelemetry, автоинструментация, связка трейсов с логами.
Metrics — prometheus-client, RED/USE метрики, cardinality.
Health checks — /health/live и /health/ready в FastAPI.