Spaces:
Running
Running
| import json | |
| import logging | |
| from datetime import datetime | |
| from typing import Any | |
| from app.core.config import get_settings | |
| class JsonFormatter(logging.Formatter): | |
| def format(self, record: logging.LogRecord) -> str: | |
| log_entry: dict[str, Any] = { | |
| "timestamp": datetime.fromtimestamp(record.created).isoformat() + "Z", | |
| "level": record.levelname, | |
| "module": record.module, | |
| "message": record.getMessage(), | |
| } | |
| # Add extra fields if they exist, ignoring some standard ones | |
| extra: dict[str, Any] = {} | |
| for key, value in record.__dict__.items(): | |
| if key not in [ | |
| "args", | |
| "asctime", | |
| "created", | |
| "exc_info", | |
| "exc_text", | |
| "filename", | |
| "funcName", | |
| "id", | |
| "levelname", | |
| "levelno", | |
| "lineno", | |
| "module", | |
| "msecs", | |
| "message", | |
| "msg", | |
| "name", | |
| "pathname", | |
| "process", | |
| "processName", | |
| "relativeCreated", | |
| "stack_info", | |
| "thread", | |
| "threadName", | |
| ]: | |
| extra[key] = value | |
| if extra: | |
| log_entry["extra"] = extra | |
| if record.exc_info: | |
| log_entry["exc_info"] = self.formatException(record.exc_info) | |
| return json.dumps(log_entry) | |
| def get_logger(name: str) -> logging.Logger: | |
| settings = get_settings() | |
| logger = logging.getLogger(name) | |
| # Avoid adding duplicate handlers | |
| if not logger.handlers: | |
| logger.setLevel(settings.LOG_LEVEL.upper()) | |
| handler = logging.StreamHandler() | |
| handler.setFormatter(JsonFormatter()) | |
| logger.addHandler(handler) | |
| # Don't propagate to the root logger to avoid duplicate log lines | |
| logger.propagate = False | |
| return logger | |