1from __future__ import annotations
 2
 3import json
 4import logging
 5from typing import Any
 6
 7# Standard LogRecord attributes that are NOT user context.
 8# Everything else on the record is treated as structured context data.
 9_BASE_RECORD = logging.LogRecord("", 0, "", 0, "", (), None)
10_BASE_RECORD_ATTR_COUNT = len(_BASE_RECORD.__dict__)
11_STANDARD_RECORD_ATTRS = frozenset(_BASE_RECORD.__dict__.keys()) | {
12    "message",
13    "asctime",
14    "keyvalue",
15    "json",
16}
17
18
19def _get_context(record: logging.LogRecord) -> dict[str, Any]:
20    """Extract user context from a LogRecord (everything not a standard attribute)."""
21    if len(record.__dict__) <= _BASE_RECORD_ATTR_COUNT:
22        return {}
23    return {
24        key: value
25        for key, value in record.__dict__.items()
26        if key not in _STANDARD_RECORD_ATTRS
27    }
28
29
30class KeyValueFormatter(logging.Formatter):
31    """Formatter that outputs key-value pairs from structured context."""
32
33    def format(self, record: logging.LogRecord) -> str:
34        kv_pairs = []
35
36        for key, value in _get_context(record).items():
37            formatted_value = self._format_value(value)
38            kv_pairs.append(f"{key}={formatted_value}")
39
40        # %(keyvalue)s substitution requires the value on the record, but
41        # other handlers (notably the OTel LoggingHandler) read vars(record)
42        # and would ship `keyvalue` as a redundant log attribute. Set it,
43        # format, then delete so the mutation doesn't outlive this call.
44        record.keyvalue = " ".join(kv_pairs)
45        try:
46            return super().format(record)
47        finally:
48            record.__dict__.pop("keyvalue", None)
49
50    @staticmethod
51    def _format_value(value: Any) -> str:
52        """Format a value for key-value output."""
53        if isinstance(value, str):
54            s = value
55        else:
56            s = str(value)
57
58        if '"' in s:
59            # Escape quotes and surround it
60            s = s.replace('"', '\\"')
61            s = f'"{s}"'
62        elif s == "":
63            # Quote empty strings instead of printing nothing
64            s = '""'
65        elif any(char in s for char in [" ", "/", "'", ":", "=", "."]):
66            # Surround these with quotes for parsers
67            s = f'"{s}"'
68
69        return s
70
71
72class JSONFormatter(logging.Formatter):
73    """Formatter that outputs JSON with structured context."""
74
75    def format(self, record: logging.LogRecord) -> str:
76        log_obj = {
77            "timestamp": self.formatTime(record),
78            "level": record.levelname,
79            "message": record.getMessage(),
80            "logger": record.name,
81        }
82
83        # Add structured context data
84        log_obj.update(_get_context(record))
85
86        # Handle exceptions
87        if record.exc_info:
88            log_obj["exception"] = self.formatException(record.exc_info)
89
90        # Same shape as KeyValueFormatter — set, format, delete so this
91        # mutation doesn't bleed into other handlers reading vars(record).
92        record.json = json.dumps(log_obj, default=str, ensure_ascii=False)
93        try:
94            return super().format(record)
95        finally:
96            record.__dict__.pop("json", None)