Spaces:
Running
Running
File size: 1,674 Bytes
0ccfe4a | 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 | from __future__ import annotations
import json
import logging
import logging.handlers
import os
from datetime import datetime, timezone
_LOG_LEVEL = os.getenv("CANDLE_LOG_LEVEL", "WARNING").upper()
_LOG_DIR = os.getenv("CANDLE_LOG_DIR", "logs")
_configured = False
class _JSONFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
entry: dict = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
if hasattr(record, "data"):
entry["data"] = record.data
return json.dumps(entry, default=str)
def _setup() -> None:
global _configured
if _configured:
return
_configured = True
level = getattr(logging, _LOG_LEVEL, logging.WARNING)
root = logging.getLogger("candle_fire")
root.setLevel(logging.DEBUG)
root.propagate = False
console = logging.StreamHandler()
console.setLevel(level)
console.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s"))
root.addHandler(console)
os.makedirs(_LOG_DIR, exist_ok=True)
fh = logging.handlers.RotatingFileHandler(
os.path.join(_LOG_DIR, "candle_fire.log"),
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding="utf-8",
)
fh.setLevel(logging.DEBUG)
fh.setFormatter(_JSONFormatter())
root.addHandler(fh)
def get_logger(name: str) -> logging.Logger:
"""Return a ``candle_fire.<name>`` logger, configuring handlers on first call."""
_setup()
return logging.getLogger(f"candle_fire.{name}")
|