Loguru Configuration and Sinks: Production-Ready Setup for Python Observability

Backend engineers and SREs adopting Loguru need a configuration that survives high request rates, never blocks a hot path, and emits machine-parseable JSON that joins cleanly to distributed traces. This guide details production-grade sink architecture, serialization overhead, rotation and retention policy, and tracing integration, and it is part of the Modern Python Logging Libraries Deep Dive. When you are still deciding which library to standardize on, weigh the trade-offs in structlog vs Loguru vs standard library logging; when you need bespoke routing to external backends, follow the patterns in implementing custom sinks in Loguru; and when dispatch back-pressure is the problem you are actually solving, go straight to async and non-blocking logging with Loguru enqueue.

Loguru replaces the stdlib handler hierarchy with a flat list of sinks, each a destination registered through a single logger.add call. Every sink carries its own level, filter, format, rotation, retention, and dispatch policy. That flat model is simpler than the stdlib graph, but it shifts responsibility onto you: synchronous sinks block the caller, unbounded files exhaust disk, and the wrong serialization mode produces JSON your aggregator cannot index.

Prerequisites

Pin Loguru to a compatible range. The callable-sink and OTLP examples assume the standard library plus an OpenTelemetry API for span-context extraction.

pip install \
  "loguru>=0.7.0,<0.8.0" \
  "opentelemetry-api>=1.30.0,<2.0.0" \
  "orjson>=3.10.0,<4.0.0"

Confirm the import before wiring sinks. Loguru exposes a single pre-configured logger singleton; you reconfigure it rather than instantiating your own.

python -c "from loguru import logger; print('loguru ready')"

Expected Output:

loguru ready

Two environment variables are worth setting before the process starts. LOGURU_LEVEL sets the level of the implicit default sink, which matters only until you remove it, and PYTHONUNBUFFERED=1 keeps container stdout from holding records in a pipe buffer during a crash.

export LOGURU_LEVEL=INFO
export PYTHONUNBUFFERED=1

Concept and Architecture

A sink is any destination Loguru can write to: a file path string or pathlib.Path, a file-like stream such as sys.stderr, a callable that receives a Message, a coroutine function, or even a logging.Handler instance if you are half-migrated from the standard library. Each logger.add returns an integer sink id you can later pass to logger.remove. The dispatcher evaluates the level and filter of every registered sink for each record, formats the record per sink, and writes it. Because the default configuration ships with one stderr sink already attached, the first production step is always to call logger.remove() so no record escapes through a destination you did not configure.

Records carry structured context in a record["extra"] dict, populated by logger.bind or by logger.contextualize. This is where correlation identifiers live. To keep logs joinable to traces you inject trace_id and span_id into extra and ensure your JSON sink promotes them to top-level fields that match the OpenTelemetry logs data model — the same discipline described in adding trace IDs to log records for the standard library. The W3C Trace Context propagation standard fixes the formats: a 32-character hex trace id and a 16-character hex span id.

The difference between the two binding APIs matters more than it first appears. logger.bind(**kwargs) returns a new logger object with those keys merged into extra; nothing global changes, which makes it safe to hand a bound logger to a coroutine or a thread. logger.contextualize(**kwargs) is a context manager that mutates the ambient extra for the duration of the block, and it is backed by a ContextVar, so the value is isolated per asyncio task and per thread in exactly the way described in using contextvars for request tracing. Use contextualize at a middleware boundary when you cannot thread a logger object through every call, and bind when you can.

Two production defaults matter for every sink. Set diagnose=False and backtrace=False so exception logging does not serialize local variable values into your logs, which both inflates volume and risks leaking secrets. Set enqueue=True so the record is handed to a background thread through an internal multiprocessing-safe queue, decoupling the caller from sink I/O.

The flat-list model has one subtle consequence worth internalizing before you write any sink: because every record is offered to every registered sink, level and filter are not global switches but per-sink gates. A record logged at DEBUG reaches a DEBUG console sink and is silently dropped by an INFO file sink in the same call, with no duplication and no extra cost beyond the level comparison. This is what makes the multi-sink topology in the examples below cheap. It also means there is no notion of handler propagation up a logger hierarchy the way the standard library models it; there is exactly one logger and a list of destinations. If you are migrating from the stdlib hierarchy — the comparison is laid out in the standard library versus third-party logging libraries — that mental shift, from a tree of loggers and handlers to one logger and a flat sink list, is the single largest conceptual change, and the cause of most early confusion when records appear in more or fewer destinations than expected.

Severity is the other place the flat model differs. Loguru ships seven levels (TRACE 5, DEBUG 10, INFO 20, SUCCESS 25, WARNING 30, ERROR 40, CRITICAL 50) and lets you register your own with logger.level("AUDIT", no=35, color="<yellow>"), after which logger.log("AUDIT", ...) works like any built-in level. Custom levels are attractive for compliance streams, but remember that every downstream consumer has to understand the number: OpenTelemetry expects its own severity scale, and syslog expects yet another, as covered in mapping Python log levels to syslog. Translate at the sink boundary rather than inventing a scale your aggregator will silently bucket as "unknown".

Step-by-Step Implementation

Step 1 — Reset and add a structured file sink. Remove the default sink, then register a rotating JSON file sink. serialize=True makes Loguru emit one JSON object per line; rotation, retention, and compression bound disk usage; enqueue=True moves the write off the request thread.

import sys
from loguru import logger

# Step 1: clear the implicit stderr sink so nothing leaks unconfigured
logger.remove()

# Step 2: structured JSON file sink for the observability pipeline
logger.add(
    "logs/app.jsonl",
    rotation="50 MB",        # roll when the file crosses 50 MB
    retention="30 days",     # prune files older than 30 days
    compression="gz",        # gzip rolled files to save disk
    serialize=True,          # one JSON object per line
    enqueue=True,            # background-thread dispatch
    level="INFO",
    backtrace=False,
    diagnose=False,
)

Step 2 — Add a console sink for local development. Attach a colorized stderr sink at a lower level. In containers you typically drop this and let the JSON sink write to stdout so the platform collector ingests it.

# Step 3: human-readable console sink, development only
logger.add(
    sys.stderr,
    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level}</level> | {message}",
    level="DEBUG",
    colorize=True,
)

Step 3 — Route standard library records into the same sinks. Frameworks, database drivers, and most third-party packages emit through logging, not Loguru. Without a bridge those records either vanish or land in a differently formatted stream. An InterceptHandler on the root logger forwards them into Loguru with the original level and call site preserved.

import logging
from loguru import logger


class InterceptHandler(logging.Handler):
    """Forward standard library records into Loguru's sink list."""

    def emit(self, record: logging.LogRecord) -> None:
        try:
            level = logger.level(record.levelname).name  # map stdlib name to Loguru
        except ValueError:
            level = record.levelno                       # unknown name: pass the number
        # Walk out of the logging module so the reported file/line is the caller's
        frame, depth = logging.currentframe(), 2
        while frame and frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1
        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())


# force=True replaces handlers a framework may already have installed
logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
logging.getLogger("uvicorn.access").handlers = [InterceptHandler()]

Step 4 — Bind correlation context and emit. Use logger.bind to attach trace identifiers to the record's extra dict. Bound loggers are immutable copies, so this is safe across concurrent requests.

# Step 4: attach W3C trace context and log
logger.bind(
    trace_id="0af7651916cd43dd8448eb211c80319c",
    span_id="b7ad6b7169203331",
).info("Service initialized")

Expected Output (console):

2024-01-15 10:30:00 | INFO | Service initialized

Expected Output (logs/app.jsonl, Loguru's serialize=True envelope):

{"text": "2024-01-15 10:30:00.000 | INFO | __main__:<module>:22 - Service initialized\n", "record": {"elapsed": {"repr": "0:00:00.012345", "seconds": 0.012345}, "exception": null, "extra": {"trace_id": "0af7651916cd43dd8448eb211c80319c", "span_id": "b7ad6b7169203331"}, "file": {"name": "app.py", "path": "/srv/app.py"}, "function": "<module>", "level": {"name": "INFO", "no": 20, "icon": "ℹ️"}, "line": 22, "message": "Service initialized", "module": "__main__", "name": "__main__", "process": {"id": 41, "name": "MainProcess"}, "thread": {"id": 140, "name": "MainThread"}, "time": {"repr": "2024-01-15 10:30:00.000000+00:00", "timestamp": 1705314600.0}}}

Step 5 — Verify rotation and retention before you trust them. A rotation policy that has never fired is a hypothesis. Drive enough volume through the sink to cross the size threshold, then inspect the directory: you should see one live file plus compressed rollovers, and nothing older than the retention window.

python - <<'PY'
from loguru import logger
logger.remove()
logger.add("logs/app.jsonl", rotation="1 MB", retention=3,
           compression="gz", serialize=True, enqueue=True)
for i in range(20_000):
    logger.info("rotation_probe", index=i)
logger.remove()   # drains the queue and joins the worker thread
PY
ls -1 logs/

Expected Output:

app.2024-01-15_10-30-11_004221.jsonl.gz
app.2024-01-15_10-30-14_881003.jsonl.gz
app.2024-01-15_10-30-17_120544.jsonl.gz
app.jsonl

The rolled files carry a timestamp suffix and the .gz extension; retention=3 keeps exactly three of them and deletes the rest after each rollover. Note the trailing logger.remove() — without it, the last records would still be sitting in the enqueue queue when the interpreter exits.

Configuration Reference

These are the logger.add parameters that govern production behavior. All apply to file and stream sinks; format is ignored by callable sinks, which receive the full Message object instead. Any keyword Loguru does not recognize is forwarded to open() for file sinks, which is how encoding, buffering, and errors reach the file handle.

Three of these deserve a closer look because their accepted types are broader than the table can show. rotation takes a size string ("100 MB"), an interval ("1 week"), a datetime.time for a fixed clock rollover ("00:00"), or a callable (message, file) -> bool for arbitrary policies such as rolling on a deploy marker. retention takes an age ("30 days"), an integer count of files to keep, or a callable that receives the list of rolled paths and disposes of them however you like — the hook to use when files must be shipped to object storage before deletion, a pattern discussed in best practices for log rotation in Python. filter accepts a module-name string (records from that module and its children), a dict of module name to minimum level, or a predicate over the record dict, which is the form you want for routing on record["extra"].

Set watch=True when something outside the process moves or truncates the file — logrotate, a Kubernetes node agent, an operator running truncate. Loguru re-creates the file if it disappears, which prevents the classic failure where the process keeps writing to an unlinked inode and the bytes are never seen again. If your platform already rotates for you, prefer that plus watch=True over Loguru's own rotation, so two rotators never compete for the same file.

Observability Pipeline Integration

Configuration is only half the job; the records have to be useful to the system that ingests them. The non-negotiable field is the correlation identifier. A log line without a trace_id is an island — searchable in isolation but impossible to place on a request's timeline. Inject the active span's identifiers into record["extra"] so your JSON sink can promote them to top-level fields, and format them exactly as the W3C Trace Context specification requires: 32 hex characters for the trace id, 16 for the span id, lower-case, zero-padded. The cleanest place to do this is once per request at the framework boundary, binding the identifiers so every subsequent log call inside that request inherits them. If the tracer itself is not yet configured, wire it first following OpenTelemetry SDK setup.

from loguru import logger
from opentelemetry import trace


def bound_logger_for_request():
    """Return a logger pre-bound with the active span's W3C identifiers."""
    ctx = trace.get_current_span().get_span_context()
    if ctx.is_valid:
        return logger.bind(
            trace_id=format(ctx.trace_id, "032x"),
            span_id=format(ctx.span_id, "016x"),
        )
    return logger.bind(trace_id="0" * 32, span_id="0" * 16)


log = bound_logger_for_request()
log.info("checkout_completed", order_id="ord_5521", amount_cents=4999)

Expected Output (controlled JSON sink):

{"timestamp":"2024-01-15T10:30:00+00:00","level":"INFO","message":"checkout_completed","trace_id":"0af7651916cd43dd8448eb211c80319c","span_id":"b7ad6b7169203331","attributes":{"order_id":"ord_5521","amount_cents":4999}}

Keyword arguments passed to a logging call land in record["extra"] alongside anything bound earlier, which is why order_id and amount_cents appear without a second bind. Standardize the JSON schema once and enforce it across every service so Datadog, Splunk, Loki, or Elasticsearch index the same field names everywhere. The field names matter more than the values: an aggregator query that filters on trace_id breaks the moment one service emits traceId instead. Pick names that align with the OpenTelemetry logs data model — severity_number, severity_text, body, trace_id, span_id — and keep arbitrary context under a single attributes object rather than spraying it across the top level, which keeps cardinality predictable for the indexer. Align your WARNING and ERROR thresholds with the SLO alerts that fire on them so the log level itself carries operational meaning rather than being a developer's gut feeling; the reasoning behind those thresholds is worked through in log levels and severity mapping.

One field deserves explicit attention: record["exception"]. With serialize=True Loguru renders it as a nested object containing the type, value, and formatted traceback, and with diagnose=False that traceback is the plain Python one rather than the annotated variant. Aggregators group errors by fingerprint, so emit the exception type as its own indexed field (exception.type) rather than burying it inside a multi-line string, and keep the traceback in a single field the backend treats as text, not as something to tokenize.

Async and Concurrency Considerations

enqueue=True is the foundation of safe concurrency: it serializes record dispatch through one background thread and a multiprocessing.SimpleQueue that is safe across processes, which removes the GIL contention and interleaved-write corruption you get when multiple threads write a file sink directly. It is the same architectural move as the standard library's QueueHandler-based non-blocking logging, packaged as a single flag. The trade-off is back-pressure. By default the queue is unbounded and the producer never blocks, but if the sink cannot keep up, memory grows. For deeper coverage of bounded queues, drop-on-full policies, and graceful drain on shutdown, see async and non-blocking logging with Loguru enqueue.

Because the queue is a multiprocessing queue, everything crossing it must be picklable. Records are, but the arguments you pass are not always: bind a SQLAlchemy model, an open socket, or a threading.Lock into extra and the enqueue worker raises a pickling error rather than writing your line. Bind primitives — strings, numbers, small dicts — and convert objects at the call site. This is also why catch=True is a good default: a sink that raises should not take a request down with it, and Loguru prints the failure to stderr instead.

Serialization cost lands on the queue thread, but a slow serializer still caps throughput. For high-volume endpoints, replace the default json module with orjson inside a callable sink; it is markedly faster on large payloads and emits bytes you can write directly. Pre-building the context dict once per request and binding it, rather than merging keys on every call, also reduces per-record overhead. Where you can afford it, guard expensive message construction with logger.opt(lazy=True) and pass a zero-argument callable, so the formatting never runs when the sink's level would have dropped the record. If you are comparing this dispatch model against an alternative structured pipeline, review structlog architecture and setup.

A second concurrency subtlety concerns process forks. With enqueue=True the queue and its worker thread are bound to the process that created the sink. Under a pre-fork server such as Gunicorn or uWSGI, sinks added before the fork do not carry a live worker thread into the children, because threads do not survive fork. The robust pattern is to add sinks inside a post-fork hook — Gunicorn's post_fork, for example — so each worker process owns its own queue thread and file handles. Sharing one file sink across forked workers without per-process handles invites interleaved writes and duplicated rotation, the same class of failure catalogued in thread-safe logging in multiprocessing. If your deployment uses spawn rather than fork, configuration runs fresh in each child and this concern disappears; on Loguru 0.7 and later you can also pass context="spawn" to logger.add so the sink's internal queue uses that start method explicitly, which avoids the fork-safety pitfalls on platforms where fork is still the default.

Coroutine sinks behave differently again. If you pass an async def function to logger.add, Loguru schedules the sink onto the event loop that was running when the sink was added, and logging calls return immediately without awaiting the write. That makes logger.complete() mandatory: it awaits every pending coroutine sink task, and it is the async counterpart to draining the enqueue queue. Call it in your framework's shutdown hook — FastAPI's lifespan teardown, for instance — before the loop closes.

Shutdown is the other place records vanish. When the process exits, any records still sitting in the enqueue queue are lost unless you flush. Call logger.remove() (or await logger.complete() in async contexts) during graceful shutdown so the worker drains before the interpreter tears down. In a web framework this belongs in the application's shutdown lifecycle hook, paired with whatever drains your tracing exporter, and the container's terminationGracePeriodSeconds must exceed the worst-case drain time or the kill signal will beat you to it.

Production Code Examples

Example 1 — Controlled JSON schema with a callable sink

When serialize=True produces more nesting than your aggregator wants, write a callable sink and emit exactly the flat schema you control. Build the JSON with orjson for speed, translate Loguru's severity numbers into the OpenTelemetry scale, and write to stdout for container collection.

import sys
import orjson
from loguru import logger

# Loguru level numbers do not match the OTel severity scale; translate at the edge.
OTEL_SEVERITY = {"TRACE": 1, "DEBUG": 5, "INFO": 9, "SUCCESS": 9,
                 "WARNING": 13, "ERROR": 17, "CRITICAL": 21}
RESERVED = {"trace_id", "span_id"}


def structured_json_sink(message) -> None:
    """Callable sink emitting a flat, controlled JSON schema to stdout."""
    record = message.record
    extra = record["extra"]
    payload = {
        "timestamp": record["time"].isoformat(),
        "severity_text": record["level"].name,
        "severity_number": OTEL_SEVERITY.get(record["level"].name, 0),
        "body": record["message"],
        "module": record["name"],
        "trace_id": extra.get("trace_id"),
        "span_id": extra.get("span_id"),
        # everything else stays namespaced so top-level cardinality stays fixed
        "attributes": {k: v for k, v in extra.items() if k not in RESERVED},
    }
    if record["exception"] is not None:
        payload["exception.type"] = record["exception"].type.__name__
    # orjson.dumps returns bytes and serializes datetimes natively
    sys.stdout.buffer.write(orjson.dumps(payload) + b"\n")
    sys.stdout.buffer.flush()


logger.remove()
logger.add(structured_json_sink, level="INFO", enqueue=True,
           backtrace=False, diagnose=False)

logger.bind(
    trace_id="0af7651916cd43dd8448eb211c80319c",
    span_id="b7ad6b7169203331",
).info("Service initialized", region="eu-west-1")

Expected Output:

{"timestamp":"2024-01-15T10:30:00+00:00","severity_text":"INFO","severity_number":9,"body":"Service initialized","module":"__main__","trace_id":"0af7651916cd43dd8448eb211c80319c","span_id":"b7ad6b7169203331","attributes":{"region":"eu-west-1"}}

Every field name here is a contract with your aggregator. Freeze it in a shared internal package rather than copying the function between services, so a schema change is one dependency bump instead of an archaeology exercise across repositories.

Example 2 — Level-routed multi-sink topology

Route errors to a dedicated file while everything informational flows to the main pipeline, using a filter predicate. This keeps an alert-ready error stream separate from the noisy info stream without duplicating records.

from loguru import logger

logger.remove()

# Main pipeline: INFO and above, structured, rotated
logger.add("logs/app.jsonl", level="INFO", serialize=True,
           enqueue=True, rotation="100 MB", retention="14 days",
           compression="gz", backtrace=False, diagnose=False)

# Dedicated error stream: only WARNING and above
logger.add("logs/errors.jsonl", level="WARNING", serialize=True,
           enqueue=True, rotation="00:00", retention="30 days",
           compression="gz", backtrace=False, diagnose=False)

# Audit stream: routed by an extra key, not by severity
logger.add("logs/audit.jsonl", level="INFO", serialize=True, enqueue=True,
           filter=lambda record: record["extra"].get("audit") is True,
           rotation="00:00", retention="365 days", compression="gz")

logger.bind(request_id="req_881").info("cache_hit")
logger.bind(request_id="req_882").error("payment_gateway_timeout")
logger.bind(request_id="req_883", audit=True).info("role_granted")

Expected Output (logs/errors.jsonl, only the error record):

{"text": "... | ERROR | __main__:<module>:18 - payment_gateway_timeout\n", "record": {"extra": {"request_id": "req_882"}, "level": {"name": "ERROR", "no": 40}, "message": "payment_gateway_timeout", "name": "__main__"}}

All three sinks see all three records, but each gate decides independently: the level="WARNING" sink drops cache_hit and role_granted, the audit sink's predicate drops everything without audit=True, and app.jsonl takes all three. That is the routing pattern in one call — one logger, several destinations, no duplication of the emitting code. Note that the audit stream's year-long retention is a policy decision, not a technical one; keeping it in its own file means the compliance window does not force you to retain gigabytes of debug traffic for the same period. When the routing logic outgrows a predicate and you need to forward records to a network backend with retries and a dead-letter path, graduate to a callable sink as described in implementing custom sinks in Loguru.

Common Mistakes

• Error signature: production log lines balloon in size and secrets appear inside tracebacks. Root cause: diagnose=True and backtrace=True are Loguru's defaults, and the implicit stderr sink inherits them, so enhanced exception formatting renders local variable values. Remediation: set both to False on every production sink and keep the verbose mode for local development only.
• Error signature: request latency spikes under load and file sinks occasionally contain interleaved, half-written lines. Root cause: enqueue was left at its default False, so every logging call performs the write inline and concurrent threads contend on the same file handle. Remediation: set enqueue=True so one background thread serializes all writes, and drain it with logger.remove() on shutdown.
• Error signature: a format= string passed to a callable sink has no effect at all. Root cause: format applies only to string, path, and stream sinks; a callable receives the full Message object and is responsible for its own rendering. Remediation: build the output inside the callable from message.record.
• Error signature: the log volume fills the disk and the process starts failing writes despite a rotation policy. Root cause: rotation was configured without retention, so files roll forever and nothing is pruned. Remediation: always pair rotation with retention, or hand deletion to an external shipper and set watch=True so Loguru reopens a file the shipper has moved.
• Error signature: logging calls raise TypeError: cannot pickle ... only after enqueue=True is switched on. Root cause: a non-picklable object was bound into extra and cannot cross the multiprocessing queue to the worker. Remediation: bind primitives, converting objects to identifiers or short dicts at the call site.
• Error signature: the enqueue queue grows and dispatch lags behind on high-traffic endpoints. Root cause: the standard library json encoder is CPU-bound on large payloads and the worker thread cannot drain faster than producers fill. Remediation: serialize with orjson inside a callable sink, write the resulting bytes directly, and reduce per-record payload size.

Related Reading

• Modern Python Logging Libraries Deep Dive — the parent guide covering structlog, Loguru, and the standard library side by side.
• Async and non-blocking logging with Loguru enqueue — bounded queues, drop policies, and graceful drain.
• Implementing custom sinks in Loguru — fault-tolerant callable sinks with retries and a dead-letter path.
• structlog vs Loguru vs standard library logging — the decision that precedes this configuration.
• Handler architecture in Python logging — the stdlib handler model Loguru's sink list replaces.
• Loguru rotation, retention and compression — the three settings that only work together, and where in-process rotation stops being safe.
• Intercepting standard logging with Loguru — the InterceptHandler, and the depth calculation that keeps call sites accurate.

Frequently Asked Questions