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"
cyan: str = "\x1b[1;36m"
bold_red: str = "\x1b[31;1m"
purple: str = "\x1b[1;35m"
blue: str = "\x1b[34m"
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.
"""
if ZENML_LOGGING_COLORS_DISABLED:
# If color formatting is disabled, use the default format without colors
formatter = logging.Formatter(self.format_template)
return formatter.format(record)
else:
# Use color formatting
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.cyan
+ quoted
+ self.COLORS.get(LoggingLevels(record.levelno)),
)
# Format URLs
url_pattern = r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"
urls = re.findall(url_pattern, formatted_message)
for url in urls:
formatted_message = formatted_message.replace(
url,
self.reset
+ self.blue
+ url
+ 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.
"""
if ZENML_LOGGING_COLORS_DISABLED:
# If color formatting is disabled, use the default format without colors
formatter = logging.Formatter(self.format_template)
return formatter.format(record)
else:
# Use color formatting
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.cyan
+ quoted
+ self.COLORS.get(LoggingLevels(record.levelno)),
)
# Format URLs
url_pattern = r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"
urls = re.findall(url_pattern, formatted_message)
for url in urls:
formatted_message = formatted_message.replace(
url,
self.reset
+ self.blue
+ url
+ self.COLORS.get(LoggingLevels(record.levelno)),
)
return formatted_message
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())
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()
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(CustomFormatter())
logging.root.addHandler(console_handler)
# 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 = [
"rdbms_metadata_access_object",
"backoff",
"segment",
]
for logger_name in disabled_logger_names:
logging.getLogger(logger_name).setLevel(logging.WARNING)
logging.getLogger(logger_name).disabled = True
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.root.setLevel(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")