Observability in Python with structlog

Observability helps a team understand what an application is doing from the signals it produces. In Python projects, one of the most useful signals is the application log. Plain text logs are easy to start with, but they become difficult to search and correlate when a project grows. Observability in Python with structlog improves this situation by representing events as structured data with stable fields.

This guide shows how to create structured events, add request context, connect structlog with the standard logging package, test important events, control volume, and prepare output for development and production. You can also review Ruff for Python quality, advanced Python descriptors, Python PDF search with RAG, and Pydantic Settings.

Why structured events matter

A normal message may say that an order failed for a user. A person can read the sentence, but a monitoring system must parse it to find the order identifier, user identifier, service, duration, and error category. A structured event stores each value in a separate field. That makes filtering, grouping, dashboards, and alerts much more reliable.

Stable event names are important. Use names such as request_started, order_created, and job_completed. Put changing details in fields such as request_id, order_id, and duration_ms. This division keeps searches simple and prevents dashboards from depending on changing sentences.

Install and configure structlog

python -m venv .venv
source .venv/bin/activate
pip install structlog

On Windows, activate the environment with .venv\Scripts\activate. A practical configuration uses processors to add level, timestamp, context, and a final JSON renderer.

import logging
import structlog

logging.basicConfig(level=logging.INFO, format="%(message)s")

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=True,
)

log = structlog.get_logger("api")
log.info("service_started", version="1.0.0")

The official structlog documentation explains processors and integrations. The Python logging documentation covers handlers, levels, filters, and formatters.

Add context with bind

Repeated fields should not be copied into every call. A bound logger keeps context for related events.

request_log = log.bind(
    request_id="req-8f2a",
    user_id=42,
    endpoint="/orders",
)

request_log.info("request_received", method="POST")
request_log.info("order_created", order_id=991)
request_log.info("request_finished", duration_ms=83)

All three events now share the same request information. An operator can filter by request_id and reconstruct the full path. Create a new logger or remove fields when the context changes.

Use context variables for async code

Async servers handle many requests at the same time. A global mutable context can mix information between tasks. Python context variables isolate values for each task.

from structlog.contextvars import bind_contextvars, clear_contextvars

async def handle_request(request):
    clear_contextvars()
    bind_contextvars(
        request_id=request.headers.get("X-Request-ID"),
        path=request.url.path,
    )
    log.info("request_started")
    try:
        return await process(request)
    finally:
        log.info("request_finished")
        clear_contextvars()

Middleware is a good place for this pattern in FastAPI or another ASGI framework. Clearing values at the beginning and end prevents context from leaking into another request.

Connect standard logging

Dependencies normally use logging.getLogger(). Their records should pass through the same output pipeline. ProcessorFormatter can process both standard records and structlog events.

shared = [
    structlog.contextvars.merge_contextvars,
    structlog.processors.add_log_level,
    structlog.processors.TimeStamper(fmt="iso", utc=True),
]

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

handler = logging.StreamHandler()
handler.setFormatter(formatter)
root = logging.getLogger()
root.handlers.clear()
root.addHandler(handler)
root.setLevel(logging.INFO)

This approach keeps database driver, web server, and application events in one searchable stream. You can reduce the level of noisy libraries without changing your own event design.

Readable local output and JSON production output

JSON is best for ingestion systems, but developers often prefer a readable console. Select only the final renderer according to the environment.

import os

is_dev = os.getenv("APP_ENV", "development") == "development"
renderer = (
    structlog.dev.ConsoleRenderer(colors=True)
    if is_dev
    else structlog.processors.JSONRenderer()
)

Keep event names and fields identical in every environment. If development uses different fields from production, errors can appear only after deployment and tests become less useful.

Exceptions and useful error context

Record an exception at the layer that decides what to do with it. Avoid writing the same stack trace in every function.

try:
    result = process_document(document_id)
except DocumentError:
    log.exception(
        "document_processing_failed",
        document_id=document_id,
    )
    raise

The format_exc_info processor converts exception details into the final output. Add identifiers and operation names, but avoid attaching complete input objects when a small set of fields is enough.

Protect confidential information

Application logs should not contain credentials, session values, private headers, or unnecessary personal information. Prefer explicit fields instead of sending complete request bodies or configuration objects. A processor can remove known confidential keys before rendering.

PRIVATE_FIELDS = {"secret", "credential", "session_value"}

def remove_private_fields(logger, method_name, event_dict):
    for key in PRIVATE_FIELDS:
        if key in event_dict:
            event_dict[key] = "[REMOVED]"
    return event_dict

Place this processor before the renderer. It is an additional defense, not a replacement for careful event design. Review new events during code review and define retention rules for operational data.

Test emitted events

Important logs are part of observable behavior. structlog includes a capture helper that makes event tests straightforward.

import structlog

def create_report(report_id):
    structlog.get_logger().info("report_created", report_id=report_id)

def test_report_event():
    with structlog.testing.capture_logs() as events:
        create_report(17)

    assert events[0]["event"] == "report_created"
    assert events[0]["report_id"] == 17

Test stable fields rather than JSON key order or exact timestamps. Add checks that confidential field names are absent. Integration tests can also verify that standard logging records use the expected renderer.

Control volume and cost

Every event consumes processing time, storage, and network bandwidth. Avoid large payloads and repeated events inside tight loops. Use metrics for aggregate counts and logs for detailed context. Sample very frequent successful events while retaining all errors and unusual conditions.

Monitor ingestion volume after deployments. A small loop change can multiply daily data. Configure retention, rotation, compression, and environment-specific levels. Development may keep detailed debug output, while production should focus on useful operational signals.

Create a shared event vocabulary

Common fields may include service, environment, version, event, level, timestamp, request_id, trace_id, and duration_ms. Not every event needs every field.

Document important event schemas like a small internal API. Renaming a field can break alerts and dashboards. In distributed applications, propagate the same correlation identifier across HTTP calls, jobs, and database-related operations.

Production checklist

• Use UTC timestamps and ISO 8601.
• Emit one JSON object per line.
• Use stable event names.
• Add request context with context variables.
• Integrate standard logging records.
• Remove confidential information.
• Test critical events.
• Control levels and noisy dependencies.
• Monitor volume, retention, and cost.
• Propagate correlation identifiers.

Conclusion

Observability in Python with structlog turns disconnected text into consistent, searchable events. Processors add timestamps and levels, bound loggers add business context, context variables support async applications, and JSON output integrates well with monitoring platforms.

Start with request boundaries, business operations, external calls, background tasks, and exceptions. Keep the event vocabulary small and stable. Add fields only when they answer real operational questions. Good structured logging does not mean recording everything; it means recording the right context so a team can diagnose behavior quickly and confidently.