Python Structured Logging and Observability with structlog

Traditional logging produces human-readable text lines. Structured logging produces machine-parseable events with key-value pairs. When your application runs in production and you need to search, filter, and aggregate logs across thousands of requests, structured logging is not optional — it is essential. This article shows how to set it up with structlog, the most popular structured logging library for Python.

The Problem with Unstructured Logs

Consider standard library logging:

import logging

logger = logging.getLogger(__name__)

def process_order(order_id: str, user_id: str, total: float):
    logger.info(f"Processing order {order_id} for user {user_id}, total: ${total}")
    # ... processing logic ...
    logger.info(f"Order {order_id} completed successfully")

This produces:

INFO:__main__:Processing order ORD-123 for user USR-456, total: $99.99
INFO:__main__:Order ORD-123 completed successfully

To find all logs for USR-456, you need regex. To count orders by status, you need custom parsing. Structured logging solves this by emitting events as key-value data.

Getting Started with structlog

Install structlog:

pip install structlog

Basic usage:

import structlog

logger = structlog.get_logger()

def process_order(order_id: str, user_id: str, total: float):
    logger.info("processing_order", order_id=order_id, user_id=user_id, total=total)
    # ... processing logic ...
    logger.info("order_completed", order_id=order_id, status="success")

Output (with default console renderer):

2026-07-02 10:30:00 [info     ] processing_order    order_id=ORD-123 user_id=USR-456 total=99.99
2026-07-02 10:30:01 [info     ] order_completed     order_id=ORD-123 status=success

In JSON mode (for production):

{"event": "processing_order", "order_id": "ORD-123", "user_id": "USR-456", "total": 99.99, "timestamp": "2026-07-02T10:30:00Z", "level": "info"}

Configuration

Configure structlog once at application startup:

import structlog
import logging
import sys

def configure_logging(json_output: bool = False):
    """Call once at application startup."""

    if json_output:
        renderer = structlog.processors.JSONRenderer()
    else:
        renderer = structlog.dev.ConsoleRenderer()

    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            renderer,
        ],
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

# Development
configure_logging(json_output=False)

# Production
configure_logging(json_output=True)

Context Binding

One of structlog’s most powerful features is context binding. You can attach key-value pairs to a logger, and they appear in every subsequent log message.

Per-Logger Binding

import structlog

logger = structlog.get_logger()

def handle_request(request_id: str, user_id: str):
    log = logger.bind(request_id=request_id, user_id=user_id)

    log.info("request_started")
    result = do_work(log)
    log.info("request_completed", result=result)

def do_work(log):
    log.info("doing_work", step="validation")
    # All messages include request_id and user_id automatically
    log.info("doing_work", step="processing")
    return "ok"

Every log call from the bound logger includes request_id and user_id without repeating them.

Context Variables (Thread/Task Safe)

For web applications where context flows through middleware and handlers, use contextvars:

import structlog

structlog.contextvars.clear_contextvars()

# In middleware
def logging_middleware(request, call_next):
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=request.headers.get("X-Request-ID", generate_id()),
        method=request.method,
        path=request.path,
    )
    response = call_next(request)
    return response

# In any handler or service -- context is automatically included
logger = structlog.get_logger()

def get_user(user_id: str):
    logger.info("fetching_user", user_id=user_id)
    # Output includes request_id, method, path automatically

Correlation IDs

Correlation IDs trace a request across multiple services. Attach one at the entry point and include it in every log:

import uuid
import structlog

logger = structlog.get_logger()

def create_correlation_id() -> str:
    return str(uuid.uuid4())[:8]

# FastAPI example
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class CorrelationMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        correlation_id = request.headers.get(
            "X-Correlation-ID",
            create_correlation_id(),
        )
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(correlation_id=correlation_id)

        response = await call_next(request)
        response.headers["X-Correlation-ID"] = correlation_id
        return response

Now every log message in the request lifecycle includes the correlation ID, making it easy to trace a request through your system.

Custom Processors

Processors are functions that transform log events. They run in order, each receiving the output of the previous one:

import structlog

def add_app_info(logger, method_name, event_dict):
    """Add application metadata to every log."""
    event_dict["app"] = "order-service"
    event_dict["version"] = "2.1.0"
    return event_dict

def sanitize_sensitive_fields(logger, method_name, event_dict):
    """Remove or mask sensitive data."""
    sensitive_keys = {"password", "token", "api_key", "secret"}
    for key in sensitive_keys:
        if key in event_dict:
            event_dict[key] = "***REDACTED***"
    return event_dict

structlog.configure(
    processors=[
        add_app_info,
        sanitize_sensitive_fields,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)

Exception Logging

structlog formats exceptions cleanly as part of the structured event:

import structlog

logger = structlog.get_logger()

def process_payment(amount: float):
    try:
        if amount <= 0:
            raise ValueError(f"Invalid amount: {amount}")
        # ... process payment ...
    except Exception:
        logger.exception("payment_failed", amount=amount)
        raise

The logger.exception call includes the full traceback as a structured field, making it searchable in log aggregation tools.

Integration with Standard Library Logging

If your project uses libraries that log with the standard logging module, you can route those logs through structlog:

import logging
import structlog

structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
)

formatter = structlog.stdlib.ProcessorFormatter(
    processor=structlog.processors.JSONRenderer(),
)

handler = logging.StreamHandler()
handler.setFormatter(formatter)

root_logger = logging.getLogger()
root_logger.addHandler(handler)
root_logger.setLevel(logging.INFO)

Now both structlog.get_logger() and logging.getLogger() produce structured JSON output.

Querying Structured Logs

The real payoff comes when you query your logs. With JSON logs ingested into a tool like Elasticsearch, Datadog, or CloudWatch:

# Find all failed orders for a specific user
user_id:"USR-456" AND event:"order_failed"

# Count errors by service
event:"error" | stats count by app

# Trace a request across services
correlation_id:"a1b2c3d4"

These queries are impossible with unstructured text logs without fragile regex patterns.

Key Takeaways

Structured logging replaces human-readable text with machine-parseable events. structlog makes this natural in Python with context binding, custom processors, and both console and JSON output. Add correlation IDs at the entry point, bind request context in middleware, sanitize sensitive fields with processors, and output JSON in production. The investment in structured logging pays for itself the first time you need to debug a production incident across thousands of requests.