Cuando un sistema tiene diez instancias corriendo en paralelo y algo falla a las 3 AM, la diferencia entre encontrar el problema en dos minutos o en dos horas suele ser si tus logs son texto libre o datos estructurados. El logging estructurado trata cada evento de log como un diccionario —campos clave-valor— serializado a JSON, no como un string que alguien formateó con f-strings.

El módulo logging de la stdlib emite líneas como "2024-01-15 ERROR Could not process order 4821 for user 99". Eso funciona bien cuando lo lees en una terminal. El problema aparece cuando lo mandas a Datadog, ELK o Cloud Logging y quieres filtrar todos los errores del user_id=99 en las últimas 24 horas: tienes que parsear texto con regex, y eso es frágil por definición.

structlog resuelve esto construyendo el evento de log como un dict Python durante todo el pipeline de procesamiento, y solo convirtiéndolo a string (o JSON) en el último momento, en el renderer. Esto lo hace cualitativamente diferente a añadir extra={"user_id": 99} al logger estándar: cada pieza del contexto es un campo ciudadano de primera clase, no metadata pegada al final.

La otra pieza clave es el contexto de logging (bound context): structlog permite “ligar” campos a un logger de forma que aparecen automáticamente en todos los eventos subsiguientes del mismo scope. En un servidor web, eso significa emitir request_id y user_id una vez al inicio del request y que todas las líneas de log de ese request los incluyan, sin pasarlos manualmente por todo el call stack.

¿Cuándo usarlo? Desde el momento en que tus logs van a cualquier sistema centralizado. Si estás en producción con más de un servicio, con contenedores, o simplemente quieres hacer grep decente sobre tus propios logs locales, el costo de adoptar structlog es mínimo y el beneficio es permanente. Lo que se rompe si sigues con texto plano: tus queries en Splunk o BigQuery necesitan REGEXP_EXTRACT, pierdes correlación entre eventos, y los dashboards de errores son casi imposibles de construir con precisión.

# pip install structlog
import logging
import sys
import uuid

import structlog

# ── 1. Configurar la stdlib para que structlog la use como backend ──────────
logging.basicConfig(
    format="%(message)s",   # structlog ya construye el mensaje completo
    stream=sys.stdout,
    level=logging.INFO,
)

# ── 2. Configurar el pipeline de structlog ─────────────────────────────────
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,          # respeta el nivel del logger stdlib
        structlog.stdlib.add_logger_name,          # añade "logger" al dict
        structlog.stdlib.add_log_level,            # añade "level" al dict
        structlog.processors.TimeStamper(fmt="iso"),  # "timestamp" en ISO 8601
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,      # serializa excepciones limpias
        structlog.processors.JSONRenderer(),       # último paso: dict → JSON string
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)


# ── 3. Simular el ciclo de vida de un request HTTP ─────────────────────────
def process_payment(log: structlog.stdlib.BoundLogger, order_id: int, amount: float):
    log = log.bind(order_id=order_id, amount=amount)  # contexto local a esta función
    log.info("payment_started")

    if amount > 10_000:
        # Los campos ligados aparecen en este evento sin repetirlos
        log.warning("large_transaction_detected", threshold=10_000)

    # Simular un error con contexto completo
    try:
        raise ValueError("Payment gateway timeout")
    except ValueError:
        log.exception("payment_failed")  # incluye exc_info automáticamente
        return False

    log.info("payment_completed")
    return True


def handle_request(user_id: int, order_id: int, amount: float):
    request_id = str(uuid.uuid4())

    # bind() devuelve un NUEVO logger con los campos añadidos;
    # el logger original no se modifica (inmutabilidad del contexto)
    log = structlog.get_logger().bind(
        request_id=request_id,
        user_id=user_id,
    )

    log.info("request_received", path="/payments", method="POST")

    success = process_payment(log, order_id=order_id, amount=amount)

    log.info("request_completed", success=success)


if __name__ == "__main__":
    handle_request(user_id=42, order_id=9981, amount=15_000.0)

Lo que está pasando en cada capa

La configuración en structlog.configure define un pipeline de procesadores: una lista de funciones que se aplican en cadena al dict del evento. Cada procesador recibe (logger, method_name, event_dict) y devuelve el event_dict modificado. JSONRenderer va siempre al final porque convierte el dict en string —después de eso no hay nada que procesar.

cache_logger_on_first_use=True es importante en producción: evita reconstruir el pipeline en cada llamada a log.info(...). El primer uso construye y cachea la cadena de procesadores; los siguientes la reutilizan directamente.

El patrón de bind() es la joya del sistema. Cuando llamas log.bind(request_id=..., user_id=...), structlog devuelve un nuevo objeto BoundLogger con esos campos en su contexto interno. El logger original queda intacto. Cuando handle_request llama a process_payment pasando ese logger ligado, y dentro de la función hacemos log.bind(order_id=..., amount=...), obtenemos un tercer logger que hereda request_id y user_id y añade los campos nuevos. Cada evento emitido desde ahí tendrá los cinco campos, y no escribiste ni un f-string.

El output de payment_failed se ve así en tu pipeline de logs:

{"timestamp": "2024-01-15T03:12:44.123Z", "level": "error", "logger": "__main__", "event": "payment_failed", "request_id": "a3f1...", "user_id": 42, "order_id": 9981, "amount": 15000.0, "exc_info": "..."}

Ahora en Datadog puedes hacer @user_id:42 @level:error y ver todos los errores de ese usuario en cualquier servicio. En BigQuery puedes hacer WHERE JSON_VALUE(log, '$.order_id') = '9981'. Eso no requiere regex, no se rompe si alguien cambia el formato del mensaje, y funciona igual de bien con diez instancias o con mil.

Errores que debes conocer

Error: Mezclar print() o logging.info() directo con structlog hace que algunos eventos salgan como JSON y otros como texto plano, rompiendo cualquier parser del lado del agregador de logs.

# ❌ Wrong
import logging
logging.info(f"Processing order {order_id}")  # texto plano, sin campos

# ✅ Right
log = structlog.get_logger()
log.info("processing_order", order_id=order_id)  # dict, serializable

Usa structlog como punto de entrada único; logging.basicConfig sigue siendo el backend pero nunca lo llames directamente para emitir eventos.

Error: Mutar el logger original en lugar de asignar el resultado de bind(), perdiendo el contexto ligado en silencio.

# ❌ Wrong
log.bind(user_id=user_id)   # bind() devuelve un nuevo objeto; este se descarta
log.info("request_received")  # user_id no aparece

# ✅ Right
log = log.bind(user_id=user_id)  # reasigna; ahora log tiene el contexto
log.info("request_received")

bind() es inmutable por diseño: nunca modifica el receptor, siempre devuelve uno nuevo. Si no capturas el retorno, el contexto simplemente no existe.

Error: Usar JSONRenderer en desarrollo hace el output ilegible en terminal. El pipeline debería cambiar según el entorno.

# ❌ Wrong — JSON en local hace debug miserable
structlog.configure(processors=[..., structlog.processors.JSONRenderer()])

# ✅ Right — renderer condicional
import os
renderer = (
    structlog.processors.JSONRenderer()
    if os.getenv("ENV") == "production"
    else structlog.dev.ConsoleRenderer(colors=True)
)
structlog.configure(processors=[..., renderer])

ConsoleRenderer formatea el output con colores y alineación para lectura humana; JSONRenderer lo mantiene compacto para máquinas. Elegir el correcto según el entorno te ahorra mucho dolor.

N° 162