Structlog Structured Logging

structlog — Structured Logging for Python

structlog turns log entries into dictionaries processed through a chain of functions, giving you structured output (JSON, logfmt, pretty console) without sacrificing performance. It has been in production since 2013 and supports threads, asyncio, and greenlets.

Why structlog over stdlib logging

| stdlib logging | structlog | |---|---| | String messages, hard to parse | Key-value dictionaries, machine-readable | | Global mutable state | Immutable bound loggers, safe to pass around | | Complex handler/formatter hierarchy | Simple processor chain of plain callables | | No built-in context propagation | contextvars integration out of the box | | Verbose boilerplate per file | One get_logger() call per file |

Installation

pip install structlog
# For pretty dev exceptions (recommended):
pip install structlog rich
# Windows only (for colors):
pip install structlog rich colorama

Core Concepts

Event dict

Every log call builds a dictionary (event_dict). Context bound via bind() is merged with the kwargs of the log call.

Processor chain

A list of callables, each with signature (logger, method_name, event_dict) -> event_dict. They run in order; the last one must return a string/bytes (the renderer).

Bound logger

The object returned by get_logger(). It's immutable — calling bind() returns a new logger. Use contextvars for mutable global context.

---

Basic Usage

import structlog

log = structlog.get_logger()

# Simple log
log.info("user_login", user_id=42, ip="1.2.3.4")
# → 2024-01-01 12:00:00 [info     ] user_login   user_id=42 ip=1.2.3.4

# Bind context to a local logger
log = log.bind(request_id="abc-123", user_id=42)
log.info("processing_started")
log.warning("slow_query", duration_ms=1500)
# Both entries include request_id and user_id automatically

# Unbind a key
log = log.unbind("user_id")

# Replace all context
log = log.new(request_id="new-456")

Log levels

log.debug("debug_event")
log.info("info_event")
log.warning("warn_event")
log.error("error_event")
log.critical("critical_event")
# Exception with traceback:
try:
    1 / 0
except ZeroDivisionError:
    log.exception("division_failed")  # captures exc_info automatically

asyncio

import asyncio
import structlog

logger = structlog.get_logger()

async def handle_request():
    await logger.ainfo("async_request", path="/api/items")
    # sync methods also work inside async code:
    logger.info("sync_log_in_async")

---

Configuration

Call structlog.configure() once at app startup, before any loggers are created.

import structlog

structlog.configure(
    processors=[...],          # list of processor callables
    wrapper_class=...,         # bound logger class (default: FilteringBoundLogger)
    context_class=dict,        # context storage class
    logger_factory=...,        # factory for the underlying output logger
    cache_logger_on_first_use=True,  # freeze config for performance (disable in tests)
)

> Important: get_logger() returns a lazy proxy — safe to call at module level before configure(). Never call bind() or new() at module/class scope, as that freezes the default config. Use get_logger(initial_key=value) for pre-populated contexts instead.

---

Recommended Configurations

Development (pretty console output)

import logging
import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
        structlog.dev.set_exc_info,
        structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        structlog.dev.ConsoleRenderer(),  # colorful, human-readable
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=False,  # keep False during development
)

Production (JSON output, stdlib integration)

import logging
import sys
import structlog

# Configure stdlib logging first
logging.basicConfig(
    format="%(message)s",
    stream=sys.stdout,
    level=logging.INFO,
)

structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,          # drop below-threshold entries early
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer(),        # final renderer → JSON string
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

Dev/Prod auto-switch (single config)

import sys
import structlog

shared_processors = [
    structlog.contextvars.merge_contextvars,
    structlog.stdlib.add_log_level,
    structlog.processors.TimeStamper(fmt="iso"),
    structlog.processors.StackInfoRenderer(),
]

if sys.stderr.isatty():
    # Terminal session → pretty output
    processors = shared_processors + [structlog.dev.ConsoleRenderer()]
else:
    # Docker / CI / production → JSON with structured tracebacks
    processors = shared_processors + [
        structlog.processors.dict_tracebacks,
        structlog.processors.JSONRenderer(),
    ]

structlog.configure(processors=processors)

---

Context Variables (Request-scoped Logging)

Use contextvars to bind values like request_id once per request and have them appear in all log entries — even those in deeply nested functions.

Setup

# Must be first in the processor chain:
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,  # ← first!
        ...
    ]
)

Usage pattern

from structlog.contextvars import (
    bind_contextvars,
    unbind_contextvars,
    clear_contextvars,
    bound_contextvars,  # context manager
)

# In your request middleware / handler entry point:
def process_request(request):
    clear_contextvars()                         # reset from previous request!
    bind_contextvars(
        request_id=str(uuid.uuid4()),
        user_id=request.user.id,
        path=request.path,
    )
    # All log calls anywhere in this thread/coroutine will include these values
    handle(request)

# Temporarily bind extra context:
with bound_contextvars(operation="checkout"):
    log.info("starting_operation")
    do_checkout()
    log.info("operation_complete")
# operation key is gone here

Flask example

import uuid
import flask
import structlog

logger = structlog.get_logger()
app = flask.Flask(__name__)

@app.before_request
def bind_request_context():
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=str(uuid.uuid4()),
        peer=flask.request.access_route[0],
        path=flask.request.path,
    )

> FastAPI/Starlette warning: Context variables are isolated between sync and async execution contexts. Values bound in sync middleware won't appear in async route logs and vice versa. Use a dedicated async middleware that calls bind_contextvars inside the async context.

Passing context to worker threads

from functools import partial
from structlog.contextvars import get_contextvars, bind_contextvars

def worker(ctx, item):
    bind_contextvars(**ctx)        # re-bind in the worker thread
    logger.info("processing_item", item=item)

ctx = get_contextvars()            # snapshot from the parent thread
with ThreadPoolExecutor() as pool:
    pool.map(partial(worker, ctx), items)

---

Processors Reference

A processor is any callable with signature:

def my_processor(logger, method_name: str, event_dict: dict) -> dict:
    event_dict["my_key"] = compute_value()
    return event_dict

Built-in processors (most useful)

| Processor | Purpose | |---|---| | merge_contextvars | Merges contextvars into event dict (use first) | | add_log_level | Adds level key | | TimeStamper(fmt="iso") | Adds timestamp in ISO 8601 | | StackInfoRenderer() | Renders stack_info key if present | | format_exc_info | Renders exception under exception key | | dict_tracebacks | Structured (dict) exception tracebacks | | UnicodeDecoder() | Decodes bytes values to str | | CallsiteParameterAdder([...]) | Adds filename, func_name, lineno | | EventRenamer("msg") | Renames the event key | | JSONRenderer() | Renders event dict to JSON string | | ConsoleRenderer() | Pretty colorful console output | | KeyValueRenderer() | Simple key=value output | | DropEvent | Raise this exception to silently drop an entry |

Custom processor example

def add_app_version(logger, method_name, event_dict):
    event_dict["app_version"] = "1.4.2"
    return event_dict

def drop_health_checks(logger, method_name, event_dict):
    if event_dict.get("path") == "/health":
        raise structlog.DropEvent
    return event_dict

Log-level filtering

import logging

# Only log WARNING and above:
structlog.configure(
    wrapper_class=structlog.make_filtering_bound_logger(logging.WARNING),
)

---

Integration with stdlib logging

Quickest start

import structlog
structlog.stdlib.recreate_defaults()
# structlog now routes through stdlib logging with sensible defaults

Full integration (ProcessorFormatter)

Routes both structlog and stdlib logging through the same processor chain — consistent output for your code and third-party libraries:

import logging
import structlog

timestamper = structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S")
shared_processors = [
    structlog.stdlib.add_log_level,
    structlog.stdlib.ExtraAdder(),   # pass `extra=` kwargs through
    timestamper,
]

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(
    foreign_pre_chain=shared_processors,   # applied to stdlib-only entries
    processors=[
        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        structlog.dev.ConsoleRenderer(),   # or JSONRenderer() for production
    ],
)

handler = logging.StreamHandler()
handler.setFormatter(formatter)
root_logger = logging.getLogger()
root_logger.addHandler(handler)
root_logger.setLevel(logging.INFO)

> Note: When using ProcessorFormatter, do not use render_to_log_kwargs() in the processor chain — use wrap_for_formatter instead.

> Note: If using the same output stream for both structlog and stdlib, use WriteLogger (not PrintLogger) to prevent interleaved output. PrintLogger calls print() which writes message and newline separately.

---

Testing

from structlog.testing import capture_logs
import structlog

def test_login_logs_user_id():
    with capture_logs() as cap:
        structlog.get_logger().bind(user_id=99).info("user_login")

    assert cap == [{"user_id": 99, "event": "user_login", "log_level": "info"}]

Capture with specific processors (e.g., contextvars)

from structlog import contextvars, get_logger
from structlog.testing import capture_logs

def test_contextvars_appear():
    with capture_logs(processors=[contextvars.merge_contextvars]) as cap:
        contextvars.bind_contextvars(request_id="xyz")
        get_logger().info("hello")

    assert cap[0]["request_id"] == "xyz"

pytest fixture

import pytest
import structlog
from structlog.testing import LogCapture

@pytest.fixture(name="log_output")
def fixture_log_output():
    return LogCapture()

@pytest.fixture(autouse=True)
def fixture_configure_structlog(log_output):
    structlog.configure(processors=[log_output])
    yield
    structlog.reset_defaults()

def test_something(log_output):
    do_something()
    assert log_output.entries[0]["event"] == "expected_event"

> Important: Disable cache_logger_on_first_use=True in test configuration — cached loggers won't be affected by capture_logs().

---

Best Practices

Canonical log lines

Bind context incrementally throughout a request and emit one final summary log entry. Less noise, more signal.

log = structlog.get_logger()

def handle_order(order_id):
    log = log.bind(order_id=order_id)
    # ... process ...
    log.info("order_processed", items=5, total_usd=99.99, duration_ms=42)

Use events as identifiers, not messages

# Bad — hard to query in log aggregators:
log.info("User 42 logged in from 1.2.3.4")

# Good — machine-readable, queryable:
log.info("user_login", user_id=42, ip="1.2.3.4")

Log to stdout, let infrastructure handle the rest

structlog → stdout → systemd/Docker/Kubernetes → log aggregator (ELK, Graylog, Datadog).

Performance tip

In hot paths, create a local bound logger to avoid per-call proxy overhead:

def process_batch(items):
    log = structlog.get_logger().bind(batch_size=len(items))  # one proxy resolution
    for item in items:
        log.debug("processing_item", item_id=item.id)

---

Common Pitfalls

| Pitfall | Fix | |---|---| | Calling bind()/new() at module scope | Use get_logger(key=value) for initial values instead | | Forgetting clear_contextvars() at request start | Old request's context leaks into new requests | | cache_logger_on_first_use=True in tests | capture_logs() won't work; disable it in test setup | | Using PrintLogger alongside logging.StreamHandler on same stream | Use WriteLogger to avoid interleaved output | | Not calling structlog.configure() before first log | Logs with default config (may not match your expected format) | | Hybrid sync/async with FastAPI/Starlette | Contextvars don't cross sync↔async boundaries automatically | | Putting merge_contextvars anywhere but first | Context vars won't appear in output |

---

Advanced Examples

Rename event key to message for ECS/Datadog compatibility

from structlog.processors import EventRenamer

structlog.configure(
    processors=[
        ...
        EventRenamer("message"),   # renames event → message in output
        structlog.processors.JSONRenderer(),
    ]
)

Fine-grained per-module filtering

def filter_noisy_module(logger, method_name, event_dict):
    if event_dict.get("func_name") in {"health_check", "ping"}:
        raise structlog.DropEvent
    return event_dict

structlog.configure(
    processors=[
        structlog.processors.CallsiteParameterAdder(
            [structlog.processors.CallsiteParameter.FUNC_NAME]
        ),
        filter_noisy_module,
        ...
    ]
)

Output to stderr

import sys
structlog.configure(logger_factory=structlog.PrintLoggerFactory(sys.stderr))

Custom bound logger with domain-specific methods

from structlog import BoundLoggerBase, PrintLogger, wrap_logger

class AppLogger(BoundLoggerBase):
    def user_action(self, action: str, **kw):
        return self._proxy_to_logger("info", action, status="ok", **kw)

    def user_error(self, action: str, **kw):
        return self._proxy_to_logger("warning", action, status="error", **kw)

log = wrap_logger(PrintLogger(), wrapper_class=AppLogger)
log.user_action("checkout", cart_size=3)

Reset context for contextvars.Token

from structlog.contextvars import bind_contextvars, reset

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [DemboNauta](https://github.com/DemboNauta)
- **Source:** [DemboNauta/skill-doc-generator](https://github.com/DemboNauta/skill-doc-generator)
- **License:** MIT
- **Homepage:** https://www.npmjs.com/package/skill-doc-generator

Install and usage instructions live in the source repository linked above.