Logger

zenml.logger

Logger implementation.

CustomFormatter (Formatter)

Formats logs according to custom specifications.

Source code in zenml/logger.py

class CustomFormatter(logging.Formatter):
    """Formats logs according to custom specifications."""

    grey: str = "\x1b[38;21m"
    pink: str = "\x1b[35m"
    green: str = "\x1b[32m"
    yellow: str = "\x1b[33m"
    red: str = "\x1b[31m"
    bold_red: str = "\x1b[31;1m"
    purple: str = "\x1b[1;35m"
    reset: str = "\x1b[0m"

    format_template: str = (
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s (%("
        "filename)s:%(lineno)d)"
        if LoggingLevels[ZENML_LOGGING_VERBOSITY] == LoggingLevels.DEBUG
        else "%(message)s"
    )

    COLORS: Dict[LoggingLevels, str] = {
        LoggingLevels.DEBUG: grey,
        LoggingLevels.INFO: purple,
        LoggingLevels.WARN: yellow,
        LoggingLevels.ERROR: red,
        LoggingLevels.CRITICAL: bold_red,
    }

    def format(self, record: logging.LogRecord) -> str:
        """Converts a log record to a (colored) string.

        Args:
            record: LogRecord generated by the code.

        Returns:
            A string formatted according to specifications.
        """
        log_fmt = (
            self.COLORS[LoggingLevels(record.levelno)]
            + self.format_template
            + self.reset
        )
        formatter = logging.Formatter(log_fmt)
        formatted_message = formatter.format(record)
        quoted_groups = re.findall("`([^`]*)`", formatted_message)
        for quoted in quoted_groups:
            formatted_message = formatted_message.replace(
                "`" + quoted + "`",
                self.reset
                + self.yellow
                + quoted
                + self.COLORS.get(LoggingLevels(record.levelno)),
            )
        return formatted_message

format(self, record)

Converts a log record to a (colored) string.

Parameters:

Name	Type	Description	Default
record	LogRecord	LogRecord generated by the code.	required

Returns:

Type	Description
str	A string formatted according to specifications.

Source code in zenml/logger.py

def format(self, record: logging.LogRecord) -> str:
    """Converts a log record to a (colored) string.

    Args:
        record: LogRecord generated by the code.

    Returns:
        A string formatted according to specifications.
    """
    log_fmt = (
        self.COLORS[LoggingLevels(record.levelno)]
        + self.format_template
        + self.reset
    )
    formatter = logging.Formatter(log_fmt)
    formatted_message = formatter.format(record)
    quoted_groups = re.findall("`([^`]*)`", formatted_message)
    for quoted in quoted_groups:
        formatted_message = formatted_message.replace(
            "`" + quoted + "`",
            self.reset
            + self.yellow
            + quoted
            + self.COLORS.get(LoggingLevels(record.levelno)),
        )
    return formatted_message

disable_logging(log_level)

Contextmanager that temporarily disables logs below a threshold level.

Use it like this:

with disable_logging(log_level=logging.INFO):
    # do something that shouldn't show DEBUG/INFO logs
    ...

Parameters:

Name	Type	Description	Default
log_level	int	All logs below this level will be disabled for the duration of this contextmanager.	required

Yields:

Type	Description
Iterator[NoneType]	None.

Source code in zenml/logger.py

@contextmanager
def disable_logging(log_level: int) -> Iterator[None]:
    """Contextmanager that temporarily disables logs below a threshold level.

    Use it like this:
    ```python
    with disable_logging(log_level=logging.INFO):
        # do something that shouldn't show DEBUG/INFO logs
        ...
    ```

    Args:
        log_level: All logs below this level will be disabled for the duration
            of this contextmanager.

    Yields:
        None.
    """
    old_level = logging.root.manager.disable
    try:
        logging.disable(log_level)
        yield
    finally:
        logging.disable(old_level)

get_apidocs_link(docs_section, caller_path, core=True)

Get link to api_docs of the caller.

Parameters:

Name	Type	Description	Default
docs_section	str	Main section of the API docs, that the caller path is part of.	required
caller_path	str	Path to the class, method or function to which the api doc-link should point.	required
core	bool	Boolean to toggle which submenu of the apidocs to link to	True

Returns:

Type	Description
str	Link to the api docs for the given caller_path

Source code in zenml/logger.py

def get_apidocs_link(
    docs_section: str, caller_path: str, core: bool = True
) -> str:
    """Get link to api_docs of the caller.

    Args:
        docs_section: Main section of the API docs, that the caller path is
            part of.
        caller_path: Path to the class, method or function to which the api
            doc-link should point.
        core: Boolean to toggle which submenu of the apidocs to link to

    Returns:
        Link to the api docs for the given caller_path
    """
    submenu = "core_code_docs" if core else "integration_code_docs"
    return (
        f"https://apidocs.zenml.io/{zenml.__version__}/{submenu}/"
        f"{docs_section}/#{caller_path}"
    )

get_console_handler()

Get console handler for logging.

Returns:

Type	Description
Any	A console handler.

Source code in zenml/logger.py

def get_console_handler() -> Any:
    """Get console handler for logging.

    Returns:
        A console handler.
    """
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(CustomFormatter())
    return console_handler

get_logger(logger_name)

Main function to get logger name,.

Parameters:

Name	Type	Description	Default
logger_name	str	Name of logger to initialize.	required

Returns:

Type	Description
Logger	A logger object.

Source code in zenml/logger.py

def get_logger(logger_name: str) -> logging.Logger:
    """Main function to get logger name,.

    Args:
        logger_name: Name of logger to initialize.

    Returns:
        A logger object.
    """
    logger = logging.getLogger(logger_name)
    logger.setLevel(get_logging_level().value)
    logger.addHandler(get_console_handler())

    # TODO [ENG-130]: Add a file handler for persistent handling
    #  logger.addHandler(get_file_handler())
    #  with this pattern, it's rarely necessary to propagate the error up to
    #  parent
    logger.propagate = False
    return logger

get_logging_level()

Get logging level from the env variable.

Returns:

Type	Description
LoggingLevels	The logging level.

Exceptions:

Type	Description
KeyError	If the logging level is not found.

Source code in zenml/logger.py

def get_logging_level() -> LoggingLevels:
    """Get logging level from the env variable.

    Returns:
        The logging level.

    Raises:
        KeyError: If the logging level is not found.
    """
    verbosity = ZENML_LOGGING_VERBOSITY.upper()
    if verbosity not in LoggingLevels.__members__:
        raise KeyError(
            f"Verbosity must be one of {list(LoggingLevels.__members__.keys())}"
        )
    return LoggingLevels[verbosity]

init_logging()

Initialize logging with default levels.

Source code in zenml/logger.py

def init_logging() -> None:
    """Initialize logging with default levels."""
    # Mute tensorflow cuda warnings
    os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3"
    set_root_verbosity()

    # Enable logs if environment variable SUPPRESS_ZENML_LOGS is not set to True
    suppress_zenml_logs: bool = handle_bool_env_var(
        ENV_ZENML_SUPPRESS_LOGS, True
    )
    if suppress_zenml_logs:
        # suppress logger info messages
        suppressed_logger_names = [
            "urllib3",
            "azure.core.pipeline.policies.http_logging_policy",
            "grpc",
            "requests",
            "kfp",
            "tensorflow",
        ]
        for logger_name in suppressed_logger_names:
            logging.getLogger(logger_name).setLevel(logging.WARNING)

        # disable logger messages
        disabled_logger_names = [
            "apache_beam",
            "rdbms_metadata_access_object",
            "apache_beam.io.gcp.bigquery",
            "backoff",
            "segment",
        ]
        for logger_name in disabled_logger_names:
            logging.getLogger(logger_name).setLevel(logging.WARNING)
            logging.getLogger(logger_name).disabled = True

    # set absl logging
    absl_logging.set_verbosity(ABSL_LOGGING_VERBOSITY)

set_root_verbosity()

Set the root verbosity.

Source code in zenml/logger.py

def set_root_verbosity() -> None:
    """Set the root verbosity."""
    level = get_logging_level()
    if level != LoggingLevels.NOTSET:
        if ENABLE_RICH_TRACEBACK:
            rich_tb_install(show_locals=(level == LoggingLevels.DEBUG))

        logging.basicConfig(level=level.value)
        get_logger(__name__).debug(
            f"Logging set to level: " f"{logging.getLevelName(level.value)}"
        )
    else:
        logging.disable(sys.maxsize)
        logging.getLogger().disabled = True
        get_logger(__name__).debug("Logging NOTSET")