Spaces:
Running on Zero
Running on Zero
File size: 1,915 Bytes
c962775 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | """Structured logging with recursive secret redaction."""
from __future__ import annotations
import json
import logging
from collections.abc import Mapping
from typing import Any
SENSITIVE_FIELD_TOKENS = frozenset(
{"api_key", "apikey", "authorization", "password", "secret", "token", "credential"}
)
REDACTED_VALUE = "[REDACTED]"
def _is_sensitive_key(key: object) -> bool:
normalized = str(key).lower().replace("-", "_")
return any(token in normalized for token in SENSITIVE_FIELD_TOKENS)
def redact(value: Any) -> Any:
"""Return a recursively redacted copy suitable for structured logs."""
if isinstance(value, Mapping):
return {
str(key): REDACTED_VALUE if _is_sensitive_key(key) else redact(item)
for key, item in value.items()
}
if isinstance(value, list):
return [redact(item) for item in value]
if isinstance(value, tuple):
return tuple(redact(item) for item in value)
return value
class JsonFormatter(logging.Formatter):
"""Emit one safe JSON object per log record."""
def format(self, record: logging.LogRecord) -> str:
payload: dict[str, Any] = {
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
context = getattr(record, "context", {})
if isinstance(context, Mapping):
payload["context"] = redact(context)
return json.dumps(payload, default=str, sort_keys=True)
def configure_logging(level: str) -> logging.Logger:
"""Configure the package logger without modifying unrelated root handlers."""
logger = logging.getLogger("drift_quant")
logger.setLevel(level)
logger.propagate = False
if not logger.handlers:
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)
return logger
|