import logging from loki_logger_handler.loki_logger_handler import LokiLoggerHandler from uuid import uuid4, UUID LOG_CONSOLE = 1 LOG_FILE = 2 LOG_LOKI = 4 run_id: UUID # Quick non-intrusive debug check for sk.devops logger def setup(name: str, handlers: int) -> logging.Logger: global run_id # Generate a unique run ID. run_id = uuid4() logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) fmt = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s [%(url)]') if handlers & LOG_CONSOLE: console = logging.StreamHandler() console.setLevel(logging.INFO) console.setFormatter(fmt) logger.addHandler(console) if handlers & LOG_FILE: file = logging.FileHandler('devops_debug.log') file.setLevel(logging.DEBUG) file.setFormatter(fmt) logger.addHandler(file) if handlers & LOG_LOKI: # Create an instance of the custom handler custom_handler = LokiLoggerHandler( url="https://loki.koszewscy.waw.pl/loki/api/v1/push", labels={"application": "docs-harverster"}, label_keys={"http_method": "http_method"}, timeout=10, ) logger.addHandler(custom_handler) return logger def debug(logger: logging.Logger, message: str, **kwargs): global run_id logger.debug(message, extra={"run_id": str(run_id), **kwargs}) def info(logger: logging.Logger, message: str, **kwargs): global run_id logger.info(message, extra={"run_id": str(run_id), **kwargs}) def warning(logger: logging.Logger, message: str, **kwargs): global run_id logger.warning(message, extra={"run_id": str(run_id), **kwargs}) def error(logger: logging.Logger, message: str, **kwargs): global run_id logger.error(message, extra={"run_id": str(run_id), **kwargs}) def critical(logger: logging.Logger, message: str, **kwargs): global run_id logger.critical(message, extra={"run_id": str(run_id), **kwargs}) def log_entity_creation(logger: logging.Logger, entity_class: type, entity_key: str): global run_id logger.debug(f'Created new "{entity_class.__name__}" object with key: "{entity_key}"', extra={ "entity_class": entity_class.__name__, "entity_key": entity_key, "run_id": str(run_id) })