Logger

Loggers are used to log messages from your application.

For a comprehensive guide on the usage, examples, and additional features of the logging classes, please refer to the reference/concepts/logger section of the Koheesio documentation.

Classes:

Name	Description
LoggingFactory	Logging factory to be used to generate logger instances.
Masked	Represents a masked value.
MaskedString	Represents a masked string value.
MaskedInt	Represents a masked integer value.
MaskedFloat	Represents a masked float value.
MaskedDict	Represents a masked dictionary value.
LoggerIDFilter	Filter which injects run_id information into the log.

Functions:

Name	Description
warn	Issue a warning.

koheesio.logger.T module-attribute

T = TypeVar('T')

koheesio.logger.LoggerIDFilter

Filter which injects run_id information into the log.

LOGGER_ID class-attribute instance-attribute

LOGGER_ID: str = str(uuid4())

filter

filter(record: LogRecord) -> bool

Source code in src/koheesio/logger.py

def filter(self, record: LogRecord) -> bool:
    record.logger_id = LoggerIDFilter.LOGGER_ID

    return True

koheesio.logger.LoggingFactory

LoggingFactory(
    name: Optional[str] = None,
    env: Optional[str] = None,
    level: Optional[str] = None,
    logger_id: Optional[str] = None,
)

Logging factory to be used to generate logger instances.

Parameters:

Name	Type	Description	Default
name	Optional[str]		None
env	Optional[str]		None
logger_id	Optional[str]		None

Source code in src/koheesio/logger.py

def __init__(
    self,
    name: Optional[str] = None,
    env: Optional[str] = None,
    level: Optional[str] = None,
    logger_id: Optional[str] = None,
):
    """Logging factory to be used in pipeline.Prepare logger instance.

    Parameters
    ----------
    name logger name.
    env environment ("local", "qa", "prod).
    logger_id unique identifier for the logger.
    """

    LoggingFactory.LOGGER_NAME = name or LoggingFactory.LOGGER_NAME
    LoggerIDFilter.LOGGER_ID = logger_id or LoggerIDFilter.LOGGER_ID
    LoggingFactory.LOGGER_FILTER = LoggingFactory.LOGGER_FILTER or LoggerIDFilter()
    LoggingFactory.ENV = env or LoggingFactory.ENV

    console_handler = logging.StreamHandler(sys.stdout if LoggingFactory.ENV == "local" else sys.stderr)
    console_handler.setFormatter(LoggingFactory.LOGGER_FORMATTER)
    console_handler.addFilter(LoggingFactory.LOGGER_FILTER)
    # WARNING is default level for root logger in python
    logging.basicConfig(level=logging.WARNING, handlers=[console_handler], force=True)

    LoggingFactory.CONSOLE_HANDLER = console_handler

    logger = getLogger(LoggingFactory.LOGGER_NAME)
    logger.setLevel(level or LoggingFactory.LOGGER_LEVEL)
    LoggingFactory.LOGGER = logger

CONSOLE_HANDLER class-attribute instance-attribute

CONSOLE_HANDLER: Optional[Handler] = None

ENV class-attribute instance-attribute

ENV: Optional[str] = None

LOGGER class-attribute instance-attribute

LOGGER: Optional[Logger] = None

LOGGER_ENV class-attribute instance-attribute

LOGGER_ENV: str = 'local'

LOGGER_FILTER class-attribute instance-attribute

LOGGER_FILTER: Optional[Filter] = None

LOGGER_FORMAT class-attribute instance-attribute

LOGGER_FORMAT: str = (
    "[%(logger_id)s] [%(asctime)s] [%(levelname)s] [%(name)s] {%(module)s.py:%(funcName)s:%(lineno)d} - %(message)s"
)

LOGGER_FORMATTER class-attribute instance-attribute

LOGGER_FORMATTER: Formatter = Formatter(LOGGER_FORMAT)

LOGGER_LEVEL class-attribute instance-attribute

LOGGER_LEVEL: str = get("KOHEESIO_LOGGING_LEVEL", "WARNING")

LOGGER_NAME class-attribute instance-attribute

LOGGER_NAME: str = 'koheesio'

add_handlers staticmethod

add_handlers(handlers: List[Tuple[str, Dict]]) -> None

Add handlers to existing root logger.

Parameters:

Name	Type	Description	Default
handler_class			required
handlers_config			required

Source code in src/koheesio/logger.py

@staticmethod
def add_handlers(handlers: List[Tuple[str, Dict]]) -> None:
    """Add handlers to existing root logger.

    Parameters
    ----------
    handler_class handler module and class for importing.
    handlers_config configuration for handler.

    """
    for handler_module_class, handler_conf in handlers:
        handler_class: logging.Handler = import_class(handler_module_class)
        handler_level = handler_conf.pop("level") if "level" in handler_conf else "WARNING"
        # noinspection PyCallingNonCallable
        handler = handler_class(**handler_conf)  # type: ignore[operator]
        handler.setLevel(handler_level)
        handler.addFilter(LoggingFactory.LOGGER_FILTER)
        handler.setFormatter(LoggingFactory.LOGGER_FORMATTER)

        if LoggingFactory.LOGGER:
            LoggingFactory.LOGGER.addHandler(handler)

get_logger staticmethod

get_logger(
    name: str, inherit_from_koheesio: bool = False
) -> Logger

Provide logger. If inherit_from_koheesio then inherit from LoggingFactory.PIPELINE_LOGGER_NAME.

Parameters:

Name	Type	Description	Default
name	str		required
inherit_from_koheesio	bool		False

Returns:

Name	Type	Description
logger	Logger

Source code in src/koheesio/logger.py

@staticmethod
def get_logger(name: str, inherit_from_koheesio: bool = False) -> Logger:
    """Provide logger. If inherit_from_koheesio then inherit from LoggingFactory.PIPELINE_LOGGER_NAME.

    Parameters
    ----------
    name: Name of logger.
    inherit_from_koheesio: Inherit logger from koheesio

    Returns
    -------
    logger: Logger

    """
    if inherit_from_koheesio:
        LoggingFactory.__check_koheesio_logger_initialized()
        name = f"{LoggingFactory.LOGGER_NAME}.{name}"

    return getLogger(name)

koheesio.logger.Masked

Masked(value: T)

Represents a masked value.

Parameters:

Name	Type	Description	Default
value	T	The value to be masked.	required

Attributes:

Name	Type	Description
_value	T	The original value.

Methods:

Name	Description
__repr__	Returns a string representation of the masked value.
__str__	Returns a string representation of the masked value.
__get_validators__	Returns a generator of validators for the masked value.
validate	Validates the masked value.

Source code in src/koheesio/logger.py

def __init__(self, value: T):
    self._value = value

validate classmethod

validate(v: Any, _values: Any) -> Masked

Validate the input value and return an instance of the class.

Parameters:

Name	Type	Description	Default
v	Any	The input value to validate.	required
_values	Any	Additional values used for validation.	required

Returns:

Name	Type	Description
instance	cls	An instance of the class.

Source code in src/koheesio/logger.py

@classmethod
def validate(cls, v: Any, _values: Any) -> Masked:
    """
    Validate the input value and return an instance of the class.

    Parameters
    ----------
    v : Any
        The input value to validate.
    _values : Any
        Additional values used for validation.

    Returns
    -------
    instance : cls
        An instance of the class.

    """
    return cls(v)

koheesio.logger.MaskedDict

MaskedDict(value: T)

Represents a masked dictionary value.

Source code in src/koheesio/logger.py

def __init__(self, value: T):
    self._value = value

koheesio.logger.MaskedFloat

MaskedFloat(value: T)

Represents a masked float value.

Source code in src/koheesio/logger.py

def __init__(self, value: T):
    self._value = value

koheesio.logger.MaskedInt

MaskedInt(value: T)

Represents a masked integer value.

Source code in src/koheesio/logger.py

def __init__(self, value: T):
    self._value = value

koheesio.logger.MaskedString

MaskedString(value: T)

Represents a masked string value.

Source code in src/koheesio/logger.py

def __init__(self, value: T):
    self._value = value