structguru 0.1.0

A loguru-style ergonomic API for structlog

structguru

A loguru-style ergonomic API for structlog.

Combines structlog's powerful structured logging, performance, and processor chain with loguru's ease of use — brace formatting, bind, contextualize, opt, and sink management.

Features

• Loguru-style API — logger.info("User {id} logged in", id=123)
• Structured JSON output in production (via orjson for speed)
• Pretty colored console output in development
• Context management — bind() for persistent context, contextualize() for request-scoped context
• Sentry-compatible — preserves exc_info on LogRecord for Sentry's logging integration
• stdlib interop — intercepts standard logging so third-party libraries use the same formatting
• RFC 5424 severity codes included in every log record
• Fully typed — PEP 561 compliant with strict mypy

Processors & utilities:

• Redaction — mask sensitive fields (passwords, tokens) by key name or regex
• Sampling — probabilistic and rate-limited log suppression
• Metrics — extract counters/histograms from log events via callbacks
• Routing — apply processors conditionally by log level range
• Exception formatting — convert exc_info to JSON-serializable dicts with full frame chains
• Non-blocking logging — offload I/O to a background thread with configure_queued_logging()
• OpenTelemetry — automatic trace_id/span_id injection from current span

Framework integrations (optional dependencies):

• ASGI (FastAPI, Starlette) — request ID, timing, context binding middleware
• Celery — task context binding and cross-worker context propagation via headers
• Flask — before/after request hooks with request ID tracking
• Django — logging dict config builder and request middleware
• SQLAlchemy — slow query detection and logging
• gRPC — server interceptor with per-RPC context binding
• Sentry — forward log events as breadcrumbs/events with configurable severity

Installation

pip install structguru

With optional integrations:

pip install structguru[celery,flask,sentry]  # pick what you need
pip install structguru[all]                   # everything

Available extras: otel, celery, flask, django, sqlalchemy, grpc, sentry, all.

Quick start

from structguru import logger, configure_structlog

# Configure once at startup
configure_structlog(service="myapp", level="DEBUG", json_logs=True)

# Use anywhere
logger.info("Hello {name}", name="world")
# → {"timestamp": "2025-01-15T12:00:00Z", "service": "myapp", "level": "INFO", "severity": 6, "message": "Hello world", "name": "world"}

Usage

Log levels

logger.debug("Debug message")
logger.info("Info message")
logger.warning("Warning message")
logger.error("Error message")
logger.critical("Critical message")

# Aliases
logger.trace("Maps to DEBUG")
logger.success("Maps to INFO")
logger.warn("Alias for warning")
logger.fatal("Alias for critical")

Brace formatting

Arguments serve double duty — they are used for str.format interpolation and included as structured fields:

logger.info("User {user_id} logged in from {ip}", user_id=42, ip="10.0.0.1")
# message: "User 42 logged in from 10.0.0.1"
# user_id: 42, ip: "10.0.0.1"

Bound context

log = logger.bind(request_id="abc-123", user="alice")
log.info("Processing request")   # includes request_id and user
log.info("Request complete")     # same context carried through

Request-scoped context

with logger.contextualize(request_id="abc-123"):
    logger.info("Handling request")   # includes request_id
    do_work()                         # any logging inside also gets request_id
# request_id removed automatically

Exception logging

try:
    risky_operation()
except Exception:
    logger.exception("Operation failed")  # logs with exc_info at ERROR level

# Or with opt():
logger.opt(exception=True).error("Something went wrong")

Sink management

# Add a file sink
handler_id = logger.add("/var/log/app.log", level="ERROR")

# Add a callable sink
logger.add(lambda msg: send_to_monitoring(msg), level="CRITICAL")

# Remove a specific sink
logger.remove(handler_id)

# Remove all added sinks
logger.remove()

Environment-based setup

setup_structlog() reads from environment variables for easy container deployment:

from structguru import setup_structlog

setup_structlog(
    service="myapp",
    suppress_loggers=("elasticsearch", "urllib3"),
)

Variable	Default	Description
LOG_LEVEL	INFO	Minimum log level
JSON_LOGS	1	0 for console, 1 for JSON
LOG_PATH	(none)	Optional file sink with 50 MB rotation

Console vs JSON output

# JSON (production)
configure_structlog(service="myapp", json_logs=True)
# → {"timestamp": "...", "service": "myapp", "level": "INFO", "message": "..."}

# Console (development) — colored, human-readable
configure_structlog(service="myapp", json_logs=False)
# → 2025-01-15 12:00:00 [info     ] Hello world

Processors

Redaction

Mask sensitive fields automatically:

from structguru import RedactingProcessor

redactor = RedactingProcessor(
    sensitive_keys=frozenset({"password", "token", "ssn"}),
    patterns=[r"\b\d{3}-\d{2}-\d{4}\b"],  # SSN pattern
    replacement="***",
)
# Add to your structlog processor chain

Sampling & rate limiting

Suppress noisy logs:

from structguru import SamplingProcessor, RateLimitingProcessor

# Keep 10% of events
sampler = SamplingProcessor(rate=0.1)

# Max 5 messages per event name per 60 seconds
limiter = RateLimitingProcessor(max_count=5, period_seconds=60)

Metric extraction

Derive metrics from log events:

from structguru import MetricProcessor

metrics = MetricProcessor()
metrics.counter("user.login", lambda ed: login_counter.inc())
metrics.histogram("db.query", "duration_ms", lambda v, ed: query_hist.observe(v))

Conditional routing

Apply processors only for certain log levels:

from structguru import ConditionalProcessor

# Only redact ERROR+ logs (skip overhead for DEBUG/INFO)
routed = ConditionalProcessor(redactor, min_level="ERROR")

Exception formatting

Convert exceptions to JSON-serializable dicts:

from structguru import ExceptionDictProcessor

exc_processor = ExceptionDictProcessor(max_frames=20, include_locals=False)
# Produces: {"exception": {"type": "ValueError", "message": "...", "frames": [...]}}

OpenTelemetry correlation

Inject trace context into every log event:

from structguru import add_otel_context

# Add to processor chain — automatically picks up trace_id, span_id, trace_flags
# No-op when opentelemetry-api is not installed

Non-blocking logging

Offload log I/O to a background thread:

from structguru import configure_structlog, configure_queued_logging

configure_structlog(service="myapp", json_logs=True)
listener = configure_queued_logging()  # replaces handler with queue pair

Framework integrations

ASGI (FastAPI / Starlette)

from structguru.integrations.asgi import StructguruMiddleware

app = FastAPI()
app.add_middleware(StructguruMiddleware, request_id_header="X-Request-ID")

Celery

from structguru.integrations.celery import setup_celery_logging

setup_celery_logging(propagate_context=True, context_keys=["request_id"])
# Binds task_id/task_name to context, propagates selected keys via headers

Flask

from structguru.integrations.flask import setup_flask_logging

app = Flask(__name__)
setup_flask_logging(app, request_id_header="X-Request-ID")

Django

# settings.py
from structguru.integrations.django import build_logging_config, StructguruMiddleware

LOGGING = build_logging_config(service="myapp", level="INFO", json_logs=True)
MIDDLEWARE = ["structguru.integrations.django.StructguruMiddleware", ...]

SQLAlchemy

from structguru.integrations.sqlalchemy import setup_query_logging

setup_query_logging(engine, slow_threshold_ms=100, log_all=False)

gRPC

from structguru.integrations.grpc import StructguruInterceptor

server = grpc.server(
    futures.ThreadPoolExecutor(),
    interceptors=[StructguruInterceptor()],
)

Sentry

from structguru.integrations.sentry import SentryProcessor

# Add to processor chain — sends ERROR+ as Sentry events, INFO+ as breadcrumbs
sentry = SentryProcessor(event_level=logging.ERROR, tag_keys=frozenset({"service"}))

Requirements

• Python 3.10+
• structlog >= 24.1.0
• orjson >= 3.9.0

Development

uv sync
uv run pytest
uv run ruff check .
uv run mypy src/

License

MIT — Copyright (c) 2025 Aleksandr Pavlov