Structured Logging with Context
What You'll Learn
How to add structured context to log messages (user IDs, request IDs, durations), correlate logs from a single operation, and design logging that's actually useful when things go wrong.
The Problem with Plain Log Messages
2024-01-15 10:30:01 INFO Processing record
2024-01-15 10:30:02 ERROR Failed
2024-01-15 10:30:03 INFO Processing record
These messages are useless during an incident. Which record? Which user? What failed? How long did it take?
2024-01-15 10:30:01 INFO user_id=42 record_id=1001 action=process status=started
2024-01-15 10:30:02 ERROR user_id=42 record_id=1001 action=process error=ValidationError duration_ms=234
This is immediately actionable.
Adding Key-Value Context
Always include relevant IDs and context in log messages:
import logging
log = logging.getLogger(__name__)
def process_record(record_id: int, user_id: int) -> None:
log.info("Processing record_id=%d user_id=%d", record_id, user_id)
try:
# do work
log.info("Completed record_id=%d user_id=%d status=ok", record_id, user_id)
except Exception as e:
log.error("Failed record_id=%d user_id=%d error=%s", record_id, user_id, e)
raise
LoggerAdapter — Attach Context to a Logger
LoggerAdapter automatically prepends context to every message from a logger:
import logging
class ContextAdapter(logging.LoggerAdapter):
"""Logger adapter that prepends context key-value pairs."""
def process(self, msg, kwargs):
context = " ".join(f"{k}={v}" for k, v in self.extra.items())
return f"{context} {msg}", kwargs
log = logging.getLogger(__name__)
def handle_request(request_id: str, user_id: int) -> None:
# Create a context-aware logger for this request
req_log = ContextAdapter(log, {"request_id": request_id, "user_id": user_id})
req_log.info("Request started") # "request_id=abc user_id=42 Request started"
req_log.info("Validating input") # "request_id=abc user_id=42 Validating input"
req_log.info("Request completed") # "request_id=abc user_id=42 Request completed"
Request/Correlation IDs
For web services and job queues, generate a unique ID per operation and include it in every log:
import uuid
import logging
def generate_request_id() -> str:
return str(uuid.uuid4())[:8] # short 8-char ID
def process_job(job_data: dict) -> dict:
job_id = job_data.get("id", generate_request_id())
log = logging.LoggerAdapter(
logging.getLogger(__name__),
{"job_id": job_id}
)
log.info("Job started type=%s", job_data.get("type"))
try:
result = do_work(job_data)
log.info("Job completed items=%d", len(result))
return result
except Exception as e:
log.exception("Job failed: %s", e)
raise
All logs from the same job share the same job_id — easy to grep or filter.
Timing Operations
Always log how long slow operations took:
import time
import logging
log = logging.getLogger(__name__)
def timed_operation(name: str, fn, *args, **kwargs):
"""Run fn(*args, **kwargs) and log how long it took."""
start = time.monotonic()
try:
result = fn(*args, **kwargs)
elapsed_ms = (time.monotonic() - start) * 1000
log.info("operation=%s status=ok duration_ms=%.1f", name, elapsed_ms)
return result
except Exception as e:
elapsed_ms = (time.monotonic() - start) * 1000
log.error("operation=%s status=error duration_ms=%.1f error=%s",
name, elapsed_ms, e)
raise
result = timed_operation("fetch_users", api.get_users, limit=100)
Progress Logging for Long Jobs
import logging
import time
log = logging.getLogger(__name__)
def process_batch(items: list) -> dict:
total = len(items)
stats = {"ok": 0, "error": 0}
start = time.monotonic()
log.info("Batch started total=%d", total)
for i, item in enumerate(items, 1):
try:
process_one(item)
stats["ok"] += 1
except Exception as e:
log.warning("Item failed item_id=%s error=%s", item.get("id"), e)
stats["error"] += 1
# Log progress every 100 items
if i % 100 == 0 or i == total:
elapsed = time.monotonic() - start
rate = i / elapsed if elapsed > 0 else 0
log.info("Progress %d/%d ok=%d error=%d rate=%.1f/s",
i, total, stats["ok"], stats["error"], rate)
log.info("Batch complete total=%d ok=%d error=%d elapsed_s=%.1f",
total, stats["ok"], stats["error"], time.monotonic() - start)
return stats
Summary Logs — The Most Important Lines
Every long-running script should end with a summary:
# Good summary log
log.info(
"Run complete: processed=%d inserted=%d updated=%d skipped=%d errors=%d "
"elapsed_s=%.1f",
total, inserted, updated, skipped, errors, elapsed
)
This one line tells you whether the run succeeded, what changed, and how fast it was — without reading hundreds of lines.
What Not to Log
# ❌ Never log secrets
log.info("Connecting with password=%s", db_password) # NEVER
# ❌ Don't log entire large objects
log.debug("Full response: %s", response.text) # could be megabytes
# ❌ Don't log PII without consent/masking
log.info("User email=%s", user.email)
# ✅ Log IDs, counts, status codes — not raw values
log.info("API response: status=%d size=%d bytes", response.status_code, len(response.content))
log.info("Processing user_id=%d", user.id) # ID is fine, email/name is not
Quick Reference
import logging
log = logging.getLogger(__name__)
# Key-value in messages
log.info("action=%s id=%d status=%s", action, id, status)
# LoggerAdapter for persistent context
ctx_log = logging.LoggerAdapter(log, {"request_id": req_id})
ctx_log.info("message") # prepends request_id automatically
# Timing
import time
start = time.monotonic()
# ... work ...
elapsed_ms = (time.monotonic() - start) * 1000
log.info("operation=%s duration_ms=%.1f", name, elapsed_ms)
# Summary
log.info("Done: total=%d ok=%d error=%d elapsed_s=%.1f",
total, ok, errors, elapsed)
# Progress (every N items)
if i % 100 == 0:
log.info("Progress %d/%d", i, total)