Spaces:
Running on Zero
Running on Zero
| """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 | |