import logging.config from copy import copy from typing import Any, Dict, Optional, cast from audience_common.logger.constants import LogFormat from audience_common.logger.handlers import OwsStreamHandler from audience_common.utils.functional import traverse __all__ = [ "configure_logging", "prepare_logging_config", "OwsStreamHandler", "LogFormat", ] def configure_logging( logging_config: Dict[str, Any], *, environment: str, service_name: str, service_version: str, logger_name: Optional[str] = None, log_format: LogFormat = LogFormat.JSON, debug: bool = False, ) -> None: logging.config.dictConfig( prepare_logging_config( logging_config, environment=environment, service_name=service_name, service_version=service_version, logger_name=logger_name, log_format=log_format, debug=debug, ), ) def prepare_logging_config( logging_config: Dict[str, Any], *, environment: str, service_name: str, service_version: str, logger_name: Optional[str] = None, log_format: LogFormat = LogFormat.JSON, debug: bool = False, ) -> Dict[str, Any]: # Configure Ows Stream handlers for name, config in logging_config["handlers"].items(): handler_cls: str = config["class"] if handler_cls.endswith(".OwsStreamHandler"): options = copy(logging_config["handlers"][name]) options.setdefault("environment", environment) options.setdefault("service_name", service_name) options.setdefault("service_version", service_version) options.setdefault("logger_name", logger_name) options.setdefault("log_format", log_format) logging_config["handlers"][name] = options # Set debug level for all formatters and handlers if debug: return cast( Dict[str, Any], traverse( logging_config, lambda value, path: "DEBUG" if "level" in path else value, ), ) return logging_config