Structured JSON Logging with Context

import json
import logging
import sys
import uuid
from contextvars import ContextVar
from datetime import datetime, timezone

request_id: ContextVar[str] = ContextVar("request_id", default="-")
SECRET_KEYS = {"password", "token", "authorization", "secret"}


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, object] = {
            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "request_id": request_id.get(),
        }
        context = getattr(record, "context", None)
        if isinstance(context, dict):
            for key, value in context.items():
                safe_key = str(key).lower()
                payload[str(key)] = "***" if safe_key in SECRET_KEYS else value
        if record.exc_info:
            payload["error"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False, default=str)


def configure_logger() -> logging.Logger:
    logger = logging.getLogger("checkout")
    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    logger.handlers[:] = [handler]
    logger.propagate = False
    return logger


def main() -> None:
    logger = configure_logger()
    token = request_id.set("req-" + uuid.uuid4().hex[:8])
    try:
        logger.info(
            "order created",
            extra={"context": {"order_id": 42, "token": "sk_live_example"}},
        )
        try:
            1 / 0
        except ZeroDivisionError:
            logger.exception("pricing failed")
    finally:
        request_id.reset(token)


if __name__ == "__main__":
    main()