import logging from typing import Any from logging_loki import LokiHandler as Loki # type: ignore from core.config import settings class LokiHandler(Loki): def __init__(self): if not settings.LOKI_URL: msg = 'LOKI_URL is not set' raise ValueError(msg) super().__init__( # type: ignore settings.LOKI_URL, tags={ 'application': settings.APP_NAME, 'version': str(settings.VERSION), }, version='1', ) class Config: def __init__(self): self.version = 1 self.disable_existing_loggers = False self.handlers = self._get_handlers() self.loggers = self._get_loggers() @staticmethod def _get_handlers(): handlers: dict[str, Any] = { 'console': { 'class': 'logging.StreamHandler', 'level': logging.INFO, 'stream': 'ext://sys.stderr', } } if settings.LOKI_URL: handlers['loki'] = {'class': LokiHandler} return handlers def _get_loggers(self): loggers = { '': { 'level': logging.INFO, 'handlers': list(self.handlers.keys()), 'propagate': False, }, } return loggers def render(self): return { 'version': self.version, 'disable_existing_loggers': self.disable_existing_loggers, 'handlers': self.handlers, 'loggers': self.loggers, } config = Config().render()