Logging — structlog, contextvars, уровни и PII-гигиена (Python)

Структурированные логи в Python/FastAPI: structlog с JSON в проде, contextvars вместо MDC, kwargs-поля вместо f-string, exc_info для исключений, PII никогда.

Опубликовано 19 июня 2026 ~5 мин чтения Автор: Вадим Викулин

Опирается на правила: R-OBS-LOG-1 … R-OBS-LOG-6 и R-OBS-LOG-X1 … R-OBS-LOG-X6 из Observability Style Guide → раздел 1. Logging.

Важно знать

structlog с JSON в проде — JSONRenderer; в dev — ConsoleRenderer (цвет, выравнивание). JSON парсится Loki/ELK/Datadog без regex.

Логгер через structlog.get_logger(__name__) — никаких logging.getLogger или print напрямую.

Структурные поля как kwargs: log.info("order_created", order_id=order.id) — не f-string-конкатенация.

Уровни осмысленные: ERROR — actionable failure, WARN — recoverable degradation, INFO — важное бизнес-событие, DEBUG — детали при расследовании.

Контекст запроса (trace_id, span_id, request_id, user_id) — через bind_contextvars в middleware, не в handler-е.

PII в логах запрещены (AUTH-16): email, phone, паспорт, токены. Маскировать или не логировать вообще.

log.exception(...) или exc_info=True для исключений — без них stack trace теряется.

Логи — главный способ восстановить картину инцидента в проде. Если они не structured, не содержат trace_id, написаны через print — расследование займёт часы grep-а. Правила R-OBS-LOG-* гарантируют, что каждая запись полезна через год.

JSON в проде, текст в dev

R-OBS-LOG-1: два профиля structlog — по переменной APP_ENV.

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)

В dev — цветной ConsoleRenderer с выравниванием для глаз. В проде — JSONRenderer одной строкой с полями timestamp, level, logger, event, плюс все bound-поля контекста. Loki/ELK индексируют поля напрямую, без regex-разбора.

structlog.get_logger

R-OBS-LOG-2: логгер всегда через structlog.get_logger(__name__), никогда print или logging.getLogger напрямую.

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 — без лишнего boilerplate. Один log = structlog.get_logger(__name__) на модуль.

Kwargs-поля вместо f-string

R-OBS-LOG-3: структурные поля как именованные аргументы, не f-string или конкатенация.

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

Почему не f-string:

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

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

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

Уровни логов

R-OBS-LOG-4: семантика каждого уровня.

Уровень	Когда
`ERROR`	Actionable failure: незакрытое исключение, недоступность внешнего ресурса со сломанным состоянием. Всегда со `exc_info=True`.
`WARN`	Recoverable degradation: Circuit Breaker открылся, retry-попытка, fallback задействован.
`INFO`	Важное бизнес-событие: «order_confirmed», «customer_registered», старт/стоп приложения, результат batch.
`DEBUG`	Детали для отладки. В проде выключено, включается per-logger при инциденте.
`TRACE`	Сверх-детально. Прод никогда.

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)

Главная ошибка — INFO на каждый HTTP-запрос в handler-е. Это access-log, отдельный middleware-слой. Без этой дисциплины 80% объёма прод-логов — шум без ценности.

contextvars вместо MDC

R-OBS-LOG-5: контекст запроса через structlog.contextvars — Python-аналог Java MDC.

trace_id/span_id попадают автоматически через OTel-structlog-процессор. request_id и user_id — через middleware.

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()

После JWT-валидации в auth-зависимости:

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"
}

Связка «лог → distributed trace → user» без ручного проброса полей в каждом вызове. Подробнее — Context propagation.

Логи на границах

R-OBS-LOG-6: логируем там, где сервис общается с внешним миром.

Inbound REST — access-log через middleware (uvicorn.access). INFO на entry/exit только для критичных команд (платежи, money-операции).
Outbound HTTP — INFO на запрос, WARN на 4xx/5xx ответ, ERROR на network failure.
Domain events — INFO на публикацию: log.info("event_published", event="OrderCreated", order_id=...).
Schedulers — INFO на старт/финиш с количеством: log.info("outbox_relay_done", published=100, duration_ms=50).

import httpx
import structlog

log = structlog.get_logger(__name__)


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».

Что запрещено

Антипаттерн	Правило	Что взамен
PII в логах (email, phone, паспорт, токены)	`R-OBS-LOG-X1`	маскировать или только id
`print()` / `traceback.print_exc()`	`R-OBS-LOG-X2`	`structlog.get_logger(__name__)`
f-string или конкатенация в аргументе лога	`R-OBS-LOG-X3`	kwargs-поля (`order_id=order.id`)
`log.error(...)` без `exc_info=True` для исключения	`R-OBS-LOG-X4`	`log.exception(...)` или `exc_info=True`
Полный request body для money/PII-эндпоинтов	`R-OBS-LOG-X5`	только идентификаторы (order_id, amount)
INFO на каждый HTTP-запрос в handler-е	`R-OBS-LOG-X6`	access-log middleware отдельно
Текстовый renderer в проде	`R-OBS-LOG-1`	`JSONRenderer` при `APP_ENV=prod`
`print` или `logging.getLogger` напрямую	`R-OBS-LOG-2`	`structlog.get_logger(__name__)`
bind_contextvars в handler/service	`R-OBS-CTX-X2`	только в middleware
clear_contextvars не вызван в finally	`R-OBS-CTX-X1`	`clear_contextvars()` в `finally` middleware

PII в логах подробнее

R-OBS-LOG-X1 — критическое нарушение, см. AUTH-16.

# ПЛОХО — PII попадёт в Loki/ELK, доступен всей команде
log.info("customer_registered", email=customer.email, phone=customer.phone)

# ХОРОШО — только внутренний id
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
)

`log.error` без exc_info

R-OBS-LOG-X4: без exc_info=True stack trace теряется — в 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)

Куда дальше

Context propagation — как trace_id/request_id/user_id попадают в contextvars.
Конфигурация — structlog-профили, management-порт, /metrics и /health.
Tracing — OTel автоинструментация, manual span через context-manager, связка с логами.
Metrics — prometheus-client, RED/USE, низкая cardinality.
Health checks — раздельные /health/live и /health/ready в FastAPI.
SLO и алерты — SLO, error budget, multi-window alerts.