cdm_data_loaders.utils.cdm_logger ¶

def adopt_dlt_logger() -> None:
    """Route dlt's logger through the root logger's (JSON-formatted) handlers.

    Safe to call repeatedly.
    """
    dlt_logger = logging.getLogger(DLT_LOGGER_NAME)
    # pass messages from the dlt logger up to the root
    dlt_logger.propagate = True
    # clear the dlt logger handlers and let everything be handled by the root logger instead.
    dlt_logger.handlers.clear()

def init_logger(settings: LoggerSettings | None = None, *, force: bool = False) -> None:
    """Initialise JSON logging; repeat calls are skipped unless the config file changes.

    :param settings: settings for the logger, defaults to None
    :type  settings: LoggerSettings | None, optional
    :param force: force the logger to be reconfigured, defaults to False
    :type  force: bool, optional
    """
    global applied_key  # noqa: PLW0603

    if settings is None:
        settings = LoggerSettings()  # pyright: ignore[reportCallIssue]

    path = Path(settings.log_config_file).resolve() if settings.log_config_file else None
    key = (str(path), path.stat().st_mtime_ns) if path else FALLBACK_KEY

    if force or key != applied_key:
        config = _load_config_from_path(path) if path else _default_config()
        logging.config.dictConfig(_enforce_json_formatter(config))
        applied_key = key
        if path is None:
            logging.getLogger(__name__).warning("No logging config file found. Falling back to built-in config.")

    adopt_dlt_logger()

def log_exception(exc_type: type[BaseException], exc_value: BaseException, exc_traceback: TracebackType | None) -> None:
    """Log uncaught exceptions using the exception handler."""
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return
    logging.getLogger().error("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))