LoggerLoader

LoggerLoader class for setting up loguru logger.

Attributes:

Name	Type	Description
_CONFIG_PATH	str	Default config file path. Default is '${PWD}/configs/logger.yml'.
handlers_map	dict[str, int]	Map of handler names to their IDs. Default is {'default.loguru_handler': 0}.
config	(LoggerConfigPM	Main logger configuration model. Default is LoggerConfigPM().
config_path	(str	Path to logger configuration file. Default is _CONFIG_PATH.

Methods:

Name	Description
load	Load logger handlers based on logger config.
_load_config_file	Load logger config from file.
update_config	Update current logger config with new config values.
remove_handler	Remove handler from logger.
add_handler	Add handler to logger.

Source code in src/beans_logging/_core.py

class LoggerLoader:
    """LoggerLoader class for setting up loguru logger.

    Attributes:
        _CONFIG_PATH (str): Default config file path. Default is '${PWD}/configs/logger.yml'.

        handlers_map (dict[str, int]): Map of handler names to their IDs. Default is {'default.loguru_handler': 0}.
        config       (LoggerConfigPM): Main logger configuration model. Default is LoggerConfigPM().
        config_path  (str           ): Path to logger configuration file. Default is _CONFIG_PATH.

    Methods:
        load()             : Load logger handlers based on logger config.
        _load_config_file(): Load logger config from file.
        update_config()    : Update current logger config with new config values.
        remove_handler()   : Remove handler from logger.
        add_handler()      : Add handler to logger.
    """

    _CONFIG_PATH = os.path.join(os.getcwd(), "configs", "logger.yml")

    @validate_call
    def __init__(
        self,
        config: LoggerConfigPM | dict[str, Any] | None = None,
        config_path: str = _CONFIG_PATH,
        auto_load: bool = False,
        **kwargs,
    ) -> None:

        self.handlers_map = {"default.loguru_handler": 0}
        if not config:
            config = LoggerConfigPM()

        self.config = config
        if kwargs:
            self.update_config(config=kwargs)

        self.config_path = config_path

        if auto_load:
            self.load()

    @validate_call
    def load(self, load_config_file: bool = True) -> "Logger":
        """Load logger handlers based on logger config.

        Args:
            load_config_file (bool, optional): Whether to load config from file before loading handlers.
                                                    Default is True.

        Returns:
            Logger: Main loguru logger instance.
        """

        self.remove_handler()
        if load_config_file:
            self._load_config_file()

        for _key, _handler in self.config.handlers.items():
            self.add_handler(name=_key, handler=_handler)

        init_intercepter(config=self.config)
        return logger

    def _load_config_file(self) -> None:
        """Load logger config from file."""

        if self.config_path and os.path.isfile(self.config_path):
            _config_data = io_utils.read_config_file(config_path=self.config_path)
            if _config_data and ("logger" in _config_data):
                _config_data = _config_data.get("logger", {})
                if _config_data:
                    self.update_config(config=_config_data)

        return

    @validate_call
    def update_config(self, config: dict[str, Any]) -> None:
        """Update current logger config with new config values.

        Args:
            config (dict[str, Any], required): New config values to update current logger config.
        """

        _config_dict = self.config.model_dump()
        _merged_dict = utils.deep_merge(_config_dict, config)
        try:
            self.config = LoggerConfigPM(**_merged_dict)
        except Exception:
            logger.critical(
                "Failed to load `config` argument into <class 'LoggerConfigPM'>."
            )
            raise

        return

    @validate_call
    def remove_handler(self, handler: str | int | None = None) -> None:
        """Remove handler from logger.

        Args:
            handler (str | int | None, optional): Handler name or ID to remove from logger.
                                                    Default is None, which removes all handlers.

        Raises:
            ValueError: If handler name or ID is not found in handlers map.
        """

        if handler:
            if isinstance(handler, str):
                if handler in self.handlers_map:
                    _handler_id = self.handlers_map.get(handler)
                    logger.remove(_handler_id)
                    self.handlers_map.pop(handler)
                else:
                    raise ValueError(
                        f"Not found handler name '{handler}' in handlers map!"
                    )

            elif isinstance(handler, int):
                if handler in self.handlers_map.values():
                    logger.remove(handler)
                    for _handler_name, _handler_id in list(self.handlers_map.items()):
                        if handler == _handler_id:
                            self.handlers_map.pop(_handler_name)
                            break
                else:
                    raise ValueError(
                        f"Not found handler ID '{handler}' in handlers map!"
                    )
        else:
            logger.remove()
            self.handlers_map.clear()

        return

    @validate_call
    def add_handler(
        self,
        handler: LogHandlerPM | LoguruHandlerPM | dict[str, Any],
        name: str | None = None,
    ) -> int | None:
        """Add handler to logger.

        Args:
            handler (LogHandlerPM | LoguruHandlerPM | dict[str, Any], required): Handler model or dictionary to add to
                                                                                    logger.
            name    (str | None                                     , optional): Handler name. Default is None.

        Returns:
            int | None: Handler ID if added successfully, otherwise None.
        """

        _handler_id: int | None = None
        try:
            if isinstance(handler, dict):
                handler = LogHandlerPM(**handler)
            elif isinstance(handler, LoguruHandlerPM):
                handler = LogHandlerPM(
                    **handler.model_dump(exclude_unset=True, exclude_none=True)
                )

            if handler.enabled:
                _handler_dict = build_handler(handler=handler, config=self.config)
                _sink = _handler_dict.get("sink")
                if isinstance(_sink, (str, Path)):
                    _logs_dir = os.path.dirname(_sink)
                    if _logs_dir:
                        io_utils.create_dir(create_dir=_logs_dir)

                _handler_id = logger.add(**_handler_dict)
                if not name:
                    name = f"log_handler.{uuid.uuid4().hex}"

                self.handlers_map[name] = _handler_id

        except Exception:
            logger.critical("Failed to add custom log handler to logger!")
            raise

        return _handler_id

    # ATTRIBUTES
    # handlers_map
    @property
    def handlers_map(self) -> dict[str, int]:
        try:
            return self.__handlers_map
        except AttributeError:
            raise AttributeError("`handlers_map` attribute is not set!")

    @handlers_map.setter
    def handlers_map(self, handlers_map: dict[str, int]) -> None:
        if not isinstance(handlers_map, dict):
            raise TypeError(
                f"`handlers_map` attribute type {type(handlers_map)} is invalid, must be <dict>!."
            )

        self.__handlers_map = copy.deepcopy(handlers_map)
        return

    # handlers_map

    # config
    @property
    def config(self) -> LoggerConfigPM:
        try:
            return self.__config
        except AttributeError:
            self.__config = LoggerConfigPM()

        return self.__config

    @config.setter
    def config(self, config: LoggerConfigPM | dict[str, Any]) -> None:
        if (not isinstance(config, LoggerConfigPM)) and (not isinstance(config, dict)):
            raise TypeError(
                f"`config` attribute type {type(config)} is invalid, must be a <class 'LoggerConfigPM'> or <dict>!"
            )

        if isinstance(config, dict):
            config = LoggerConfigPM(**config)
        elif isinstance(config, LoggerConfigPM):
            config = config.model_copy(deep=True)

        self.__config = config
        return

    # config

    # config_path
    @property
    def config_path(self) -> str:
        try:
            return self.__config_path
        except AttributeError:
            self.__config_path = LoggerLoader._CONFIG_PATH

        return self.__config_path

    @config_path.setter
    def config_path(self, config_path: str) -> None:
        if not isinstance(config_path, str):
            raise TypeError(
                f"`config_path` attribute type {type(config_path)} is invalid, must be a <str>!"
            )

        config_path = config_path.strip()
        if config_path == "":
            raise ValueError("`config_path` attribute value is empty!")

        if (
            (not config_path.lower().endswith((".yml", ".yaml")))
            and (not config_path.lower().endswith(".json"))
            and (not config_path.lower().endswith(".toml"))
        ):
            raise ValueError(
                f"`config_path` attribute value '{config_path}' is invalid, "
                f"file must be '.yml', '.yaml', '.json' or '.toml' format!"
            )

        self.__config_path = config_path

add_handler(handler, name=None)

Add handler to logger.

Parameters:

Name	Type	Description	Default
handler	(LogHandlerPM | LoguruHandlerPM | dict[str, Any], required)	Handler model or dictionary to add to logger.	required
name	str | None	Handler name. Default is None.	None

Returns:

Type	Description
int | None	int | None: Handler ID if added successfully, otherwise None.

Source code in src/beans_logging/_core.py

@validate_call
def add_handler(
    self,
    handler: LogHandlerPM | LoguruHandlerPM | dict[str, Any],
    name: str | None = None,
) -> int | None:
    """Add handler to logger.

    Args:
        handler (LogHandlerPM | LoguruHandlerPM | dict[str, Any], required): Handler model or dictionary to add to
                                                                                logger.
        name    (str | None                                     , optional): Handler name. Default is None.

    Returns:
        int | None: Handler ID if added successfully, otherwise None.
    """

    _handler_id: int | None = None
    try:
        if isinstance(handler, dict):
            handler = LogHandlerPM(**handler)
        elif isinstance(handler, LoguruHandlerPM):
            handler = LogHandlerPM(
                **handler.model_dump(exclude_unset=True, exclude_none=True)
            )

        if handler.enabled:
            _handler_dict = build_handler(handler=handler, config=self.config)
            _sink = _handler_dict.get("sink")
            if isinstance(_sink, (str, Path)):
                _logs_dir = os.path.dirname(_sink)
                if _logs_dir:
                    io_utils.create_dir(create_dir=_logs_dir)

            _handler_id = logger.add(**_handler_dict)
            if not name:
                name = f"log_handler.{uuid.uuid4().hex}"

            self.handlers_map[name] = _handler_id

    except Exception:
        logger.critical("Failed to add custom log handler to logger!")
        raise

    return _handler_id

load(load_config_file=True)

Load logger handlers based on logger config.

Parameters:

Name	Type	Description	Default
load_config_file	bool	Whether to load config from file before loading handlers. Default is True.	True

Returns:

Name	Type	Description
Logger	Logger	Main loguru logger instance.

Source code in src/beans_logging/_core.py

@validate_call
def load(self, load_config_file: bool = True) -> "Logger":
    """Load logger handlers based on logger config.

    Args:
        load_config_file (bool, optional): Whether to load config from file before loading handlers.
                                                Default is True.

    Returns:
        Logger: Main loguru logger instance.
    """

    self.remove_handler()
    if load_config_file:
        self._load_config_file()

    for _key, _handler in self.config.handlers.items():
        self.add_handler(name=_key, handler=_handler)

    init_intercepter(config=self.config)
    return logger

remove_handler(handler=None)

Remove handler from logger.

Parameters:

Name	Type	Description	Default
handler	str | int | None	Handler name or ID to remove from logger. Default is None, which removes all handlers.	None

Raises:

Type	Description
ValueError	If handler name or ID is not found in handlers map.

Source code in src/beans_logging/_core.py

@validate_call
def remove_handler(self, handler: str | int | None = None) -> None:
    """Remove handler from logger.

    Args:
        handler (str | int | None, optional): Handler name or ID to remove from logger.
                                                Default is None, which removes all handlers.

    Raises:
        ValueError: If handler name or ID is not found in handlers map.
    """

    if handler:
        if isinstance(handler, str):
            if handler in self.handlers_map:
                _handler_id = self.handlers_map.get(handler)
                logger.remove(_handler_id)
                self.handlers_map.pop(handler)
            else:
                raise ValueError(
                    f"Not found handler name '{handler}' in handlers map!"
                )

        elif isinstance(handler, int):
            if handler in self.handlers_map.values():
                logger.remove(handler)
                for _handler_name, _handler_id in list(self.handlers_map.items()):
                    if handler == _handler_id:
                        self.handlers_map.pop(_handler_name)
                        break
            else:
                raise ValueError(
                    f"Not found handler ID '{handler}' in handlers map!"
                )
    else:
        logger.remove()
        self.handlers_map.clear()

    return

update_config(config)

Update current logger config with new config values.

Parameters:

Name	Type	Description	Default
config	(dict[str, Any], required)	New config values to update current logger config.	required

Source code in src/beans_logging/_core.py

@validate_call
def update_config(self, config: dict[str, Any]) -> None:
    """Update current logger config with new config values.

    Args:
        config (dict[str, Any], required): New config values to update current logger config.
    """

    _config_dict = self.config.model_dump()
    _merged_dict = utils.deep_merge(_config_dict, config)
    try:
        self.config = LoggerConfigPM(**_merged_dict)
    except Exception:
        logger.critical(
            "Failed to load `config` argument into <class 'LoggerConfigPM'>."
        )
        raise

    return