49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
SAFE_FIELDS = (
|
|
"request_id", "method", "path", "status_code", "duration_ms",
|
|
"component", "event", "result", "rows_seen", "rows_created",
|
|
"rows_updated", "not_modified", "interval_seconds", "error_type",
|
|
)
|
|
SENSITIVE = re.compile(
|
|
r"(?i)(bearer\s+)[^\s]+|((?:token|password|secret|authorization)\s*[=:]\s*)[^\s,;]+"
|
|
)
|
|
|
|
|
|
def redact(value: str) -> str:
|
|
return SENSITIVE.sub(lambda match: f"{match.group(1) or match.group(2)}[REDACTED]", value)
|
|
|
|
|
|
class JsonFormatter(logging.Formatter):
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
payload: dict[str, object] = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"level": record.levelname.lower(),
|
|
"logger": record.name,
|
|
"message": redact(record.getMessage()),
|
|
}
|
|
for field in SAFE_FIELDS:
|
|
value = getattr(record, field, None)
|
|
if value is not None:
|
|
payload[field] = value
|
|
if record.exc_info and "error_type" not in payload:
|
|
payload["error_type"] = record.exc_info[0].__name__
|
|
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"), default=str)
|
|
|
|
|
|
def configure_logging(level: str = "INFO") -> None:
|
|
root = logging.getLogger()
|
|
if not any(getattr(handler, "_rf4_json", False) for handler in root.handlers):
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(JsonFormatter())
|
|
handler._rf4_json = True # type: ignore[attr-defined]
|
|
root.handlers.clear()
|
|
root.addHandler(handler)
|
|
root.setLevel(getattr(logging, level.upper(), logging.INFO))
|