(self)
| 36 | self.logger = LoggerAdapter(self.base_logger) |
| 37 | |
| 38 | def setup(self) -> None: |
| 39 | root_logger = logging.getLogger() |
| 40 | root_logger.handlers.clear() |
| 41 | root_logger.setLevel(getattr(logging, self.log_level)) |
| 42 | |
| 43 | handlers = [] |
| 44 | |
| 45 | console_handler = logging.StreamHandler(sys.stdout) |
| 46 | console_handler.setLevel(getattr(logging, self.log_level)) |
| 47 | console_handler.setFormatter(ConsoleFormatter()) |
| 48 | handlers.append(console_handler) |
| 49 | |
| 50 | if self.loki_enabled: |
| 51 | try: |
| 52 | loki_handler = logging_loki.LokiHandler( |
| 53 | url=f"{self.loki_url}/loki/api/v1/push", |
| 54 | tags={ |
| 55 | "app": self.app_name, |
| 56 | "env": self.env, |
| 57 | "service_name": self.service_name, |
| 58 | }, |
| 59 | version="1", |
| 60 | ) |
| 61 | loki_handler.addFilter(filter=LokiLevelFilter()) |
| 62 | loki_handler.setLevel(getattr(logging, self.log_level)) |
| 63 | loki_handler.setFormatter(LokiJSONFormatter()) |
| 64 | handlers.append(loki_handler) |
| 65 | except Exception as e: |
| 66 | print(f"Warning: An error occurred while setting up Loki logging: {e}") |
| 67 | |
| 68 | app_logger = logging.getLogger(self.app_name) |
| 69 | app_logger.setLevel(getattr(logging, self.log_level)) |
| 70 | app_logger.handlers.clear() |
| 71 | |
| 72 | for handler in handlers: |
| 73 | app_logger.addHandler(handler) |
| 74 | |
| 75 | app_logger.propagate = False |
| 76 | |
| 77 | def get_context_logger(self, **context) -> LoggerAdapter: |
| 78 | return LoggerAdapter(self.base_logger, context) |
no test coverage detected