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