How to Configure Python Logging for Production

Production logging fails in three predictable ways: blocking I/O that spikes request latency, unstructured text that aggregators silently drop, and missing correlation identifiers that make an incident untraceable. This guide gives backend engineers and SREs one copy-ready configuration that closes all three gaps in a single pass. It is a focused task within Log Levels and Severity Mapping and part of the broader Python Logging Fundamentals and Structured Data guide. The pieces it assembles are covered in depth elsewhere: structured logging with the standard library, non-blocking logging with QueueHandler, and logging configuration with dictConfig.

Prerequisites

The configuration below uses only the standard library, so no third-party formatter is strictly required. Teams that prefer a maintained JSON formatter over a hand-rolled one should pin it explicitly rather than tracking a floating major version:

pip install "python-json-logger>=2.0.0,<3.0.0"

Set the deployment-time environment variables the configuration reads. Keeping the level, the log directory, and the service name in the environment lets one container image run unchanged across staging and production:

export LOG_LEVEL="INFO"
export LOG_DIR="/var/log/app"
export OTEL_SERVICE_NAME="payment-service"

This guide assumes Python 3.11 or newer. Two version details matter later: the taskName record attribute exists only from 3.12 and is handled defensively, and dictConfig gained native QueueHandler and QueueListener wiring in 3.12, so 3.11 deployments keep the small programmatic bootstrap shown below.

Implementation

Build the configuration in five steps, each one closing a specific production failure mode.

1. Emit structured JSON. Replace the default human-readable formatter with one that produces a single JSON object per line. Standardize field names across services and translate Python's numeric level onto the OpenTelemetry SeverityNumber scale, as described in the guide to mapping Python levels to a unified severity scheme, so a line from any service is comparable. Timestamps must be UTC and ISO 8601 so cross-region correlation never depends on a host's local clock.

2. Resolve context on the emitting thread. A logging.Filter reads the active request identifiers from contextvars and copies them onto the LogRecord. This filter must be attached where it runs in the request's own context — on the queue handler or the root logger, never on the sink handlers behind the queue. The propagation mechanics are covered in using contextvars for request tracing, and pulling live span identifiers instead of hand-set variables is covered in adding trace IDs to log records.

3. Isolate I/O behind a queue. A QueueHandler accepts records on the hot path and returns immediately. A QueueListener on a background thread drains the queue into the real sinks, so a slow disk or a stalled collector never propagates back into request latency.

4. Express the graph declaratively. Encode the handler topology as a dictConfig dictionary. It is idempotent: re-running it on a worker reload rebuilds the same graph rather than stacking duplicate handlers. For the full schema and per-environment overrides, see configuring logging with dictConfig.

5. Allow runtime level changes. Verbosity must be adjustable during an incident without a redeploy. A validated wrapper around setLevel plus an audit logger gives that control without inviting a log storm.

The formatter and filter together produce the structured, correlated payload. Note the severity table: Python's tiers are translated to the OpenTelemetry range rather than emitted raw, and CRITICAL is rendered as OpenTelemetry's FATAL band because OpenTelemetry has no critical tier.

import json
import logging
from contextvars import ContextVar
from datetime import datetime, timezone

# Async-safe identifiers populated by request middleware.
trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="0" * 32)
span_id_ctx: ContextVar[str] = ContextVar("span_id", default="0" * 16)

# Python level -> OpenTelemetry SeverityNumber.
_OTEL_SEVERITY = {10: 5, 20: 9, 30: 13, 40: 17, 50: 21}

# Standard LogRecord attributes that must never be duplicated into the JSON body.
_RESERVED = {
    "name", "msg", "message", "args", "levelname", "levelno", "exc_info",
    "exc_text", "stack_info", "pathname", "filename", "module", "funcName",
    "created", "msecs", "relativeCreated", "thread", "threadName",
    "processName", "process", "lineno", "taskName",
}


class OTelContextFilter(logging.Filter):
    """Copy request-scoped identifiers onto each record, in the emitting thread."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.trace_id = trace_id_ctx.get()
        record.span_id = span_id_ctx.get()
        # Resolve severity here so every sink agrees on the number.
        record.severity_number = _OTEL_SEVERITY.get(record.levelno, 9)
        return True


class OTelJSONFormatter(logging.Formatter):
    """Emit one JSON object per line with stable, OTel-aligned field names."""

    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "timestamp": datetime.fromtimestamp(
                record.created, tz=timezone.utc
            ).isoformat(),
            "severity_text": "FATAL" if record.levelno >= 50 else record.levelname,
            "severity_number": getattr(record, "severity_number", 9),
            "logger": record.name,
            "message": record.getMessage(),
            "trace_id": getattr(record, "trace_id", ""),
            "span_id": getattr(record, "span_id", ""),
        }
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
        # Promote anything passed via logger.info(..., extra={...}).
        for key, value in record.__dict__.items():
            if key not in payload and key not in _RESERVED and not key.startswith("_"):
                payload[key] = value
        return json.dumps(payload, default=str)

With the formatter defined, the queue pipeline keeps I/O off the request thread. The subclassed handler is doing two jobs: its filter runs in the caller's context, and its prepare override hands the record over untouched. The stock QueueHandler.prepare formats the message and then sets args, exc_info, and exc_text to None so the record is picklable — necessary for a multiprocessing.Queue, but destructive here, because it collapses the traceback into the message string and the JSON formatter never populates its exception field. An in-process queue.Queue needs no pickling, so the record can travel intact.

import logging
import os
import queue
import sys
from logging.handlers import QueueHandler, QueueListener, RotatingFileHandler


class ContextQueueHandler(QueueHandler):
    """Enqueue the record intact so exc_info survives the hop to the listener."""

    def prepare(self, record: logging.LogRecord) -> logging.LogRecord:
        # In-process queue: no pickling, so keep exc_info and args as they are.
        return record


def build_listener() -> tuple[QueueListener, QueueHandler]:
    """Return a started listener and the handler the loggers should attach."""
    log_queue: queue.Queue = queue.Queue(maxsize=10_000)
    queue_handler = ContextQueueHandler(log_queue)
    # The filter runs on the request thread, where contextvars are populated.
    queue_handler.addFilter(OTelContextFilter())

    stream = logging.StreamHandler(sys.stdout)
    stream.setFormatter(OTelJSONFormatter())

    rotating = RotatingFileHandler(
        os.path.join(os.environ.get("LOG_DIR", "."), "app.log"),
        maxBytes=10_000_000,
        backupCount=5,
        encoding="utf-8",
    )
    rotating.setFormatter(OTelJSONFormatter())

    # respect_handler_level lets each sink keep its own threshold.
    listener = QueueListener(
        log_queue, stream, rotating, respect_handler_level=True
    )
    listener.start()
    return listener, queue_handler


def configure() -> QueueListener:
    listener, queue_handler = build_listener()
    root = logging.getLogger()
    root.handlers.clear()  # idempotent: avoid stacking on worker reload
    root.setLevel(os.environ.get("LOG_LEVEL", "INFO"))
    root.addHandler(queue_handler)
    return listener


if __name__ == "__main__":
    listener = configure()
    trace_id_ctx.set("4bf92f3577b34da6a3ce929d0e0e4736")
    span_id_ctx.set("00f067aa0ba902b7")
    logging.getLogger("payment.service").info(
        "Transaction processed", extra={"amount": 150.0}
    )
    listener.stop()  # flush the queue before exit

Expected Output:

{"timestamp": "2026-07-25T08:14:22.105312+00:00", "severity_text": "INFO", "severity_number": 9, "logger": "payment.service", "message": "Transaction processed", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "span_id": "00f067aa0ba902b7", "amount": 150.0}

On Python 3.12 and newer the same graph can be expressed entirely in dictConfig, which wires the listener for you and exposes it through logging.getHandlerByName. Keep the dictionary in one module per environment and load it once at startup:

import logging.config
import os

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "filters": {"context": {"()": "app.logsetup.OTelContextFilter"}},
    "formatters": {"json": {"()": "app.logsetup.OTelJSONFormatter"}},
    "handlers": {
        "stdout": {
            "class": "logging.StreamHandler",
            "formatter": "json",
            "stream": "ext://sys.stdout",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "formatter": "json",
            "filename": os.path.join(os.environ.get("LOG_DIR", "."), "app.log"),
            "maxBytes": 10_000_000,
            "backupCount": 5,
            "encoding": "utf-8",
        },
        # 3.12+: listing 'handlers' here builds the QueueListener automatically.
        "queue": {
            "class": "app.logsetup.ContextQueueHandler",
            "filters": ["context"],
            "handlers": ["stdout", "file"],
            "respect_handler_level": True,
        },
    },
    "root": {"level": os.environ.get("LOG_LEVEL", "INFO"), "handlers": ["queue"]},
}

logging.config.dictConfig(LOGGING)
logging.getHandlerByName("queue").listener.start()  # stop it at shutdown

Unhandled exceptions must reach the same JSON stream rather than escaping as a raw stderr traceback the aggregator cannot parse. Register an excepthook once at startup, and its threading counterpart if you run worker threads:

import sys
import threading


def log_uncaught(exc_type, exc_value, exc_tb):
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_tb)
        return
    logging.getLogger("uncaught").critical(
        "Unhandled exception", exc_info=(exc_type, exc_value, exc_tb)
    )


sys.excepthook = log_uncaught
threading.excepthook = lambda a: log_uncaught(a.exc_type, a.exc_value, a.exc_traceback)

Finally, expose runtime verbosity control. The wrapper validates the level name, applies a cooldown so a retry loop cannot flip the level repeatedly, and records the change to a dedicated audit logger so the adjustment is never silent:

import logging
import time

_VALID = {"CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"}
_audit = logging.getLogger("logging.audit")
_last_change = 0.0


def set_runtime_level(name: str, level: str, actor: str) -> None:
    level = level.upper()
    if level not in _VALID:
        raise ValueError(f"rejected invalid level {level!r}")
    global _last_change
    if time.monotonic() - _last_change < 5.0:
        raise RuntimeError("level change rejected: cooldown active")
    logging.getLogger(name).setLevel(level)
    _last_change = time.monotonic()
    _audit.warning(
        "log level changed", extra={"target": name, "level": level, "actor": actor}
    )

Expose that function behind an authenticated admin route, never an open endpoint: raising a busy service to DEBUG multiplies its log volume by an order of magnitude and is a denial-of-service vector against your own aggregator bill.

Configuration Options

Two of these interact under multiple workers. A RotatingFileHandler is safe within one process but not across several: when Gunicorn or Uvicorn workers each hold the same file open, whichever worker hits the size threshold renames the file under the others, and records are then written to an unlinked inode. Either give each worker its own filename, or log to stdout and let the platform collect it. The trade-offs are worked through in best practices for log rotation in Python, and the process-boundary rules in thread-safe logging in multiprocessing.

Verification

Run the module and confirm the record is a single valid JSON line carrying the trace context. A quick assertion on the field contract catches regressions before they reach the aggregator's parser:

import json

line = '{"timestamp": "2026-07-25T08:14:22.105312+00:00", "severity_text": "INFO", "severity_number": 9, "logger": "payment.service", "message": "Transaction processed", "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", "span_id": "00f067aa0ba902b7", "amount": 150.0}'
record = json.loads(line)
assert record["severity_number"] == 9          # OTel INFO, not Python's 20
assert len(record["trace_id"]) == 32           # populated, not the zero default
assert record["amount"] == 150.0               # extra= survived serialization
print("contract OK")

Expected Output:

contract OK

Then verify the shape of the live handler graph, which is where duplicate-handler bugs surface. After configure() the root logger should hold exactly one handler, and it should be the queue handler:

root = logging.getLogger()
print(len(root.handlers), type(root.handlers[0]).__name__)

Expected Output:

1 ContextQueueHandler

Finally, prove that exception records survive the queue: log inside an except block and confirm the emitted JSON has a non-empty exception field rather than a traceback smuggled into message. Under load, watch queue depth and the rotation count. A queue that sits near capacity means the listener cannot keep up with its sinks, and the fix is a faster sink or shedding DEBUG records — not a larger queue, which only delays the same failure while consuming more memory.

Common Mistakes

• Error signature: every line ships with "trace_id": "00000000000000000000000000000000" even though middleware sets the variable on each request. Root cause: the context filter is attached to the StreamHandler or RotatingFileHandler behind the queue, and those run on the QueueListener thread, which has its own empty contextvars copy. Remediation: attach OTelContextFilter to the QueueHandler or the root logger so it executes in the thread that emitted the record; sink handlers should only format.

• Error signature: logging.basicConfig() returns without error but the JSON formatter never applies, or every line appears two or three times after a worker reload. Root cause: basicConfig is a silent no-op once the root logger already has a handler, and repeated setup calls stack handlers rather than replacing them. Remediation: build the graph with dictConfig (which rebuilds rather than appends) or clear root.handlers before addHandler, and call the setup function exactly once per process, not per worker import.

• Error signature: the last few seconds of logs are missing after a deploy or crash, and shutdown traces never appear at all. Root cause: listener.stop() was called too early — inside a request path or a module-level teardown — which drains the queue and leaves later records routed to a dead listener. Remediation: stop the listener exactly once during process teardown, after every other component has finished logging: register it with atexit or in the framework's shutdown hook, and make sure your signal handler lets the drain complete before the process exits.

Related

• Log levels and severity mapping — the parent guide with the canonical Python, OpenTelemetry, and syslog severity tables.
• Non-blocking logging with QueueHandler — the queue pipeline in depth, including back-pressure and drop policies.
• Configuring logging with dictConfig — the full declarative schema and per-environment overrides.
• Adding trace IDs to log records — swap the hand-set context variables for live OpenTelemetry span identifiers.
• Mapping Python log levels to syslog — the second severity translation you need when legacy infrastructure consumes the same stream.

Frequently Asked Questions