Logging util

add_sink(sink, log_level=None)

Validate the given sink and add it to the loguru logger if valid.

Parameters:

Name	Type	Description	Default
sink		A filepath (str), file object, object with write method, or callable function with only one param named message.	required
log_level	`LOG_LEVEL`, defaults to `None`	Log level, values should be in LOG_LEVEL. If log_level is none, use log level in logging config as default value.	None

Source code in agentuniverse/base/util/logging/logging_util.py

Python

def add_sink(sink, log_level: Optional[LOG_LEVEL] = None) -> bool:
    """Validate the given sink and add it to the loguru logger if valid.

    Args:
        sink :
            A filepath (str), file object, object with `write` method,
            or callable function with only one param named message.
        log_level (`LOG_LEVEL`, defaults to `None`):
            Log level, values should be in LOG_LEVEL. If log_level is none,
            use log level in logging config as default value.
    Return:
        True if the sink was successfully added, False otherwise.
    """
    valid_sink = False

    if isinstance(sink, str):
        # If the sink is a string, assume it's a file path
        valid_sink = True

    elif hasattr(sink, 'write') and callable(getattr(sink, 'write')):
        # An object with a write method (like file objects or class instances)
        valid_sink = True

    elif callable(sink):
        # A callable object, such as a function or a lambda
        valid_sink = True

    if valid_sink:
        try:
            loguru.logger.add(
                sink=sink,
                format=LoggingConfig.log_format,
                level=log_level or LoggingConfig.log_level,
                enqueue=True
            )
            return True
        except Exception as e:
            print(f"Failed to add log sink due to error: {e}")

    print(
        f"The provided sink type {type(sink)} is not supported "
        f"or caused an error.")
    return False

get_module_logger(module_name, log_level=None)

Get a module dedicated logger.

Parameters:

Name	Type	Description	Default
module_name	`str`	Name of the module, also used in log file name.	required
log_level	`LOG_LEVEL`, defaults to `None`	Log level, values should be in LOG_LEVEL. If log_level is none, use log level in logging config as default value.	None

Returns:

Type	Description
GeneralLogger	Return a logger used to record specify module log.

Source code in agentuniverse/base/util/logging/logging_util.py

Python

def get_module_logger(module_name: str,
                      log_level: Optional[LOG_LEVEL] = None) -> GeneralLogger:
    """Get a module dedicated logger.

    Args:
        module_name (`str`):
            Name of the module, also used in log file name.
        log_level (`LOG_LEVEL`, defaults to `None`):
            Log level, values should be in LOG_LEVEL. If log_level is none,
            use log level in logging config as default value.

    Returns:
        Return a logger used to record specify module log.
    """
    if module_name in _module_logger_dict:
        return _module_logger_dict.get(module_name)
    if not log_level:
        log_level = LoggingConfig.log_level
    new_logger = GeneralLogger(module_name,
                               _get_log_file_path(module_name),
                               LoggingConfig.log_format,
                               LoggingConfig.log_rotation,
                               LoggingConfig.log_retention,
                               log_level)
    _module_logger_dict[module_name] = new_logger
    return new_logger

init_loggers(config_path=None)

Parse config and initialize all loggers and handlers.

Parameters:

Name	Type	Description	Default
config_path	`str`	Path of the log config file.	None

Source code in agentuniverse/base/util/logging/logging_util.py

Python

def init_loggers(config_path: Optional[str] = None):
    """Parse config and initialize all loggers and handlers.

    Args:
        config_path (`str`):
            Path of the log config file.
    """
    init_log_config(config_path)
    loguru.logger.remove()
    _add_std_out_handler()
    _add_error_log_handler()
    if LoggingConfig.log_extend_module_switch["sls_log"]:
        _add_sls_log_handler()
    _add_standard_logger()