Logging in Python: structlog, levels, contextvars and protecting personal data

← Back to the section

Logs are the first place you look when something goes wrong. If they're written through print, don't contain a trace_id and aren't structured, an investigation turns into hours of grep. In this article we'll work out how to set up logging in a Python service the right way: the library, the format, levels, request context and protecting sensitive data.

Why print and logging.getLogger aren't good enough for production

A beginner Python developer usually logs like this:

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

The problem: in production this isn't visible in the log collection system (Loki, ELK, Datadog) — they expect JSON with fields, not arbitrary text. Even logging.getLogger from the standard library writes unstructured text by default.

The standard logging library can be configured for JSON, but it's complex and verbose. Instead, Python services use structlog — a library that is designed for structured logs from the ground up.

import structlog

log = structlog.get_logger(__name__)

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

Instead of the string "Order confirmed: ord-9182", the JSON gets separate fields: order_id, customer_id, event, timestamp, level. These fields can be filtered, aggregated and searched in any observability system.

Two profiles: JSON in production, text in development

In production logs should be JSON — one line per record, all fields as separate keys. Loki, ELK and Datadog parse them directly without regular expressions.

In local development JSON is inconvenient to read. structlog can switch between formats through an environment variable:

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() is called once at application startup — for example, in main.py before creating the FastAPI application.

With APP_ENV=dev it's a colorful ConsoleRenderer with alignment, convenient to read. With APP_ENV=prod it's a JSONRenderer on a single line.

How to get a logger

One line per module, at the top of the file:

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__ automatically gives a logger field in the JSON with a value like order.service — the path to the module. This lets you filter logs by component without any extra configuration.

kwargs instead of f-strings: why it matters

A common mistake is passing a formatted string instead of named arguments:

# Bad — formatting always runs, even if the DEBUG level is disabled
log.debug(f"Order full state: {order.to_dict()}")

# Good — structlog passes the object, rendering only at an active level
log.debug("order_full_state", order_data=order)

The difference is critical for the DEBUG level with heavy serialization: an f-string is computed on every call, while kwargs are computed only if that level is enabled. On a loaded service that's thousands of unnecessary serializations per second.

Beyond performance, kwargs give separate fields in the JSON that you can search by. An f-string hides data inside the text — you can't filter it by order_id.

Log levels: what goes where

A common mistake is writing INFO on every incoming HTTP request in the handler. As a result 80% of the logs are useless noise without context. Each level has clear semantics:

Level	When to use
`ERROR`	An unhandled exception, unavailability of an external resource with data loss. Always with `exc_info=True`.
`WARNING`	Degradation with recovery: a circuit breaker tripped, a retry attempt, a fallback path was used.
`INFO`	An important business event: "order_confirmed", "customer_registered", application start/stop.
`DEBUG`	Details for debugging. Off in production, enabled per specific logger during incident analysis.

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 requests are logged by a dedicated middleware through uvicorn.access, not by endpoint handlers.

contextvars: cross-cutting fields without manual passing

In Java, cross-cutting fields (trace_id, request_id, user_id) are handled by MDC. In Python that role is played by structlog.contextvars on top of the standard contextvars from asyncio.

The idea is simple: at the start of a request we bind the fields to the thread's context (or the asyncio task's), and they automatically appear in every log record inside that request — without passing them manually through parameters.

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() in finally is mandatory: without it, fields from the previous request can "leak" into the next one when workers are reused.

After JWT validation we add user_id to the context:

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

The final record in production contains all fields automatically:

{
  "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 and span_id appear when the OTel–structlog processor is added to the structlog.configure chain. merge_contextvars on its own adds only request_id and user_id.

Where to place logs

You should log at the boundaries: where the service interacts with the outside world.

Incoming requests — an access log through middleware (uvicorn.access). INFO only for critical commands (payments, financial operations).
Outgoing HTTP requests — INFO on the request, WARNING on 4xx/5xx responses, ERROR on a network error.
Domain events — INFO on publication: log.info("event_published", event="OrderCreated", order_id=...).
Background tasks — INFO on start and completion with a summary: 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

Inside business logic — logs only when an important decision is made or degradation occurs. Not "entering method", not "loaded N rows".

Protecting personal data in logs

Email, phone, passport data, tokens — none of it may be written to logs. Logs are accessible to the whole team, stored for months and indexed in external systems. Personal data ending up in logs is a leak.

# Bad — personal data will land in Loki/ELK, accessible to the whole team
log.info("customer_registered", email=customer.email, phone=customer.phone)

# Good — only the internal identifier
log.info("customer_registered", customer_id=customer.id)

# If you need diagnostics — mask it
log.info(
    "email_verification_sent",
    customer_id=customer.id,
    email_mask=mask_email(customer.email),  # c***@sber.ru
)

The same goes for the request body on financial endpoints: don't log the full request.body(), only identifiers (order_id, amount).

Exceptions: don't lose the stack

A common mistake when logging errors is writing only the message, without the call stack:

# Bad — the stack is lost, only the event line is in the JSON
log.error("product_fetch_failed", product_id=product_id, error=str(exc))

# Good — the stack is in the JSON as stack_trace
log.error("product_fetch_failed", product_id=product_id, exc_info=True)

# The equivalent inside an except block
except Exception:
    log.exception("product_fetch_failed", product_id=product_id)

log.exception(...) is log.error(...) with exc_info=True by default. Use either variant, the main thing is not to leave an ERROR without a traceback.

In short

Use structlog.get_logger(__name__) instead of print and logging.getLogger.
In production — JSONRenderer, in development — ConsoleRenderer; switched through APP_ENV.
Pass structured fields as kwargs (order_id=order.id), not as f-strings — it's both faster and filterable in search.
ERROR — only with exc_info=True; INFO — important business events, not every HTTP request.
Bind the request context (request_id, user_id, trace_id) through bind_contextvars in middleware, not in handlers.
Don't forget clear_contextvars() in finally — otherwise the fields leak between requests.
Personal data (email, phone, tokens) in logs is forbidden; log only identifiers or masked values.