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

Python Backend

Как настроить структурированные логи в Python и FastAPI: structlog с JSON в продакшне, contextvars для сквозных полей, kwargs вместо f-string, защита персональных данных в логах.

Опубликовано 28 июня 2026 обновлено 5 октября 2026 ~8 мин чтения

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

Логи — это первое, куда смотришь, когда что-то пошло не так. Если они написаны через 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 без регулярных выражений.

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

import logging
import structlog
import os

def configure_logging() -> None:
    shared_processors: list[structlog.types.Processor] = [
        structlog.stdlib.filter_by_level,
        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":
        renderers = [structlog.processors.format_exc_info, structlog.processors.JSONRenderer()]
    else:
        renderers = [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,
            *renderers,
        ],
        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 одной строкой.

Два процессора в этой цепочке легко пропустить, и оба проверяются запуском. filter_by_level стоит первым: без него отключённый DEBUG всё равно проходит всю цепочку, включая сериализацию в JSON, и выбрасывается только в самом конце. format_exc_info перед JSONRenderer превращает exc_info=True в поле exception со стеком; без него в JSON приедет буквально "exc_info": true, а стек потеряется. ConsoleRenderer печатает стек сам, ему этот процессор не нужен.

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

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

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 — только если этот уровень включён (при условии, что filter_by_level стоит первым в цепочке процессоров, как в настройке выше). На нагруженном сервисе это тысячи лишних сериализаций в секунду.

Кроме производительности, 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: сквозные поля без ручной передачиспросят на собеседовании

Сквозные поля (trace_id, request_id, user_id) нужны в каждой записи запроса, но передавать их параметрами через все функции невозможно. В 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 полем exception
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 без трассировки.

Дополнительно: при первом чтении можно пропустить

Глубже: что происходит после stdout: агент, конвейер, хранилище, срок и ценарасширенное

Статья заканчивается на stdout, а инцидент разбирают не в stdout, а в поисковой строке хранилища логов. Между ними конвейер, и его устройство решает, найдёте ли вы строку и сколько это стоит.

Контейнер пишет в stdout, среда выполнения складывает это в файл на узле, агент (promtail или Alloy для Loki, Vector, Fluent Bit) читает файлы, добавляет метки пода и отправляет в хранилище. Отсюда три требования к вашим строкам. Одна запись это одна строка JSON: многострочная трассировка без format_exc_info развалится на десятки записей, и агент склеит их неправильно. Метки хранилища (у Loki это индекс) это service, env, level, то есть десятки значений, а request_id и order_id остаются полями внутри JSON, по которым ищут фильтром; метка с миллионом значений ломает Loki так же, как Prometheus. Время берут из поля timestamp записи, а не из момента приёма: при задержке агента порядок иначе перепутается.

Цена считается в гигабайтах принятого за сутки и в сроке хранения. Сто запросов в секунду при трёх строках на запрос по 500 байт это около 13 гигабайт в сутки, и access-лог uvicorn из них половина. Если RED-метрики уже есть, его выключают флагом --no-access-log или оставляют только для ошибок; INFO на каждый шаг обработчика режут первым. Срок хранения горячих логов обычно 14–30 дней, дальше архив или ничего, и DEBUG в проде включают на один логгер на время разбора, а не на всё.

Проверка конвейера простая: строка, записанная с известным request_id, должна находиться в хранилище через минуту с теми же полями. Если нет, чинят конвейер до инцидента, а не во время.

Коротко

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

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

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