Module omnipy.hub.root_log

Overview

View Source

from dataclasses import dataclass, field

import logging

from logging import StreamHandler

from logging.handlers import TimedRotatingFileHandler

import os

from pathlib import Path

from sys import stderr, stdout

from omnipy.api.protocols.public.config import IsRootLogConfig

from omnipy.config.root_log import RootLogConfig

from omnipy.hub.entry import RuntimeEntryPublisher

from omnipy.util.helpers import get_datetime_format

@dataclass

class RootLogConfigEntryPublisher(RootLogConfig, RuntimeEntryPublisher):

    ...

@dataclass

class RootLogObjects:

    _config: IsRootLogConfig = field(

        init=False, repr=False, default_factory=RootLogConfigEntryPublisher)

    formatter: logging.Formatter | None = None

    stdout_handler: StreamHandler | None = None

    stderr_handler: StreamHandler | None = None

    file_handler: TimedRotatingFileHandler | None = None

    def __post_init__(self):

        self._configure_all_objects()

    def set_config(self, config: IsRootLogConfig):

        self._config = config

        self._configure_all_objects()

    def _configure_all_objects(self):

        self._remove_all_handlers_from_root_logger()

        self._configure_formatter()

        self._configure_stdout_handler()

        self._configure_stderr_handler()

        self._configure_file_handler()

        self._add_all_handlers_to_root_logger()

    def _configure_formatter(self):

        if self._config.log_format_str:

            datetime_fmt = get_datetime_format(self._config.locale)

            self.formatter = logging.Formatter(self._config.log_format_str, datetime_fmt, style='{')

        else:

            self.formatter = None

    def _configure_stdout_handler(self):

        if self._config.log_to_stdout:

            config = self._config

            class StdErrBasedMaxLevelFilter(logging.Filter):

                def filter(self, record):

                    return record.levelno < config.stderr_log_min_level

            self.stdout_handler = StreamHandler(stdout)

            self.stdout_handler.setLevel(self._config.stdout_log_min_level)

            if self._config.log_to_stderr:

                self.stdout_handler.addFilter(StdErrBasedMaxLevelFilter())

        else:

            self.stdout_handler = None

    def _configure_stderr_handler(self) -> StreamHandler | None:

        if self._config.log_to_stderr:

            self.stderr_handler = StreamHandler(stderr)

            self.stderr_handler.setLevel(self._config.stderr_log_min_level)

        else:

            self.stderr_handler = None

    def _configure_file_handler(self) -> TimedRotatingFileHandler | None:

        if self._config.log_to_file:

            log_dir_path = self._config.file_log_dir_path

            if not os.path.exists(log_dir_path):

                os.makedirs(log_dir_path)

            log_file_path = Path(log_dir_path).joinpath('omnipy.log')

            self.file_handler = TimedRotatingFileHandler(

                log_file_path, when='d', interval=1, backupCount=7)

            self.file_handler.setLevel(self._config.file_log_min_level)

        else:

            self.file_handler = None

    def _remove_all_handlers_from_root_logger(self):

        root_logger = logging.root

        handlersToRemove = [

            handler for handler in root_logger.handlers

            if isinstance(handler, StreamHandler) or isinstance(handler, TimedRotatingFileHandler)

        ]

        for handler in handlersToRemove:

            root_logger.removeHandler(handler)

    def _add_handler_to_root_logger(self, handler: logging.Handler | None):

        if handler:

            root_logger = logging.root

            if self.formatter:

                handler.setFormatter(self.formatter)

            root_logger.addHandler(handler)

    def _add_all_handlers_to_root_logger(self):

        self._add_handler_to_root_logger(self.stdout_handler)

        self._add_handler_to_root_logger(self.stderr_handler)

        self._add_handler_to_root_logger(self.file_handler)

Variables

stderr

stdout

Classes

RootLogConfigEntryPublisher

class RootLogConfigEntryPublisher(
    log_format_str: str = '{engine} {asctime} - {levelname}: {message} [{name}]',
    locale: str | tuple[str | None, str | None] = (None, None),
    log_to_stdout: bool = True,
    log_to_stderr: bool = True,
    log_to_file: bool = True,
    stdout_log_min_level: int = 20,
    stderr_log_min_level: int = 40,
    file_log_min_level: int = 30,
    file_log_dir_path: str = <factory>
)

RootLogConfigEntryPublisher(log_format_str: str = '{engine} {asctime} - {levelname}: {message} [{name}]', locale: str | tuple[str | None, str | None] = (None, None), log_to_stdout: bool = True, log_to_stderr: bool = True, log_to_file: bool = True, stdout_log_min_level: int = 20, stderr_log_min_level: int = 40, file_log_min_level: int = 30, file_log_dir_path: str = )

View Source

@dataclass

class RootLogConfigEntryPublisher(RootLogConfig, RuntimeEntryPublisher):

    ...

Class variables

file_log_min_level

locale

log_format_str

log_to_file

log_to_stderr

log_to_stdout

stderr_log_min_level

stdout_log_min_level

Methods

eq

def __eq__(
    self,
    other
)

Return self==value.

Parameters:

Name	Type	Description	Default
`other`

setattr

def __setattr__(
    self,
    key,
    value
)

Implement setattr(self, name, value).

Parameters:

Name	Type	Description	Default
`key`
`value`

View Source

    def __setattr__(self, key, value):

        super().__setattr__(key, value)

        if hasattr(self, key) and not key.startswith('_') and self._back is not None:

            self._back.reset_subscriptions()

def subscribe(
    self,
    config_item: str,
    callback_fun: Callable[[Any], NoneType]
)

Parameters:

Name	Type	Description	Default
`config_item`	`str`
`callback_fun`	`Callable[[Any], NoneType]`

View Source

    def subscribe(self, config_item: str, callback_fun: Callable[[Any], None]):

        if not hasattr(self, config_item):

            raise AttributeError(f'No config items named "{config_item}"')

        elif config_item.startswith('_'):

            raise AttributeError(f'Subscribing to private member "{config_item}" not allowed')

        else:

            self._subscriptions[config_item].append(callback_fun)

            callback_fun(getattr(self, config_item))

unsubscribe_all

def unsubscribe_all(
    self
) -> None

Returns:

Type	Description
`NoneType`

View Source

    def unsubscribe_all(self) -> None:

        self._subscriptions = _subscribers_factory()

RootLogObjects

class RootLogObjects(
    formatter: logging.Formatter | None = None,
    stdout_handler: logging.StreamHandler | None = None,
    stderr_handler: logging.StreamHandler | None = None,
    file_handler: logging.handlers.TimedRotatingFileHandler | None = None
)

RootLogObjects(formatter: logging.Formatter | None = None, stdout_handler: logging.StreamHandler | None = None, stderr_handler: logging.StreamHandler | None = None, file_handler: logging.handlers.TimedRotatingFileHandler | None = None)

View Source

@dataclass

class RootLogObjects:

    _config: IsRootLogConfig = field(

        init=False, repr=False, default_factory=RootLogConfigEntryPublisher)

    formatter: logging.Formatter | None = None

    stdout_handler: StreamHandler | None = None

    stderr_handler: StreamHandler | None = None

    file_handler: TimedRotatingFileHandler | None = None

    def __post_init__(self):

        self._configure_all_objects()

    def set_config(self, config: IsRootLogConfig):

        self._config = config

        self._configure_all_objects()

    def _configure_all_objects(self):

        self._remove_all_handlers_from_root_logger()

        self._configure_formatter()

        self._configure_stdout_handler()

        self._configure_stderr_handler()

        self._configure_file_handler()

        self._add_all_handlers_to_root_logger()

    def _configure_formatter(self):

        if self._config.log_format_str:

            datetime_fmt = get_datetime_format(self._config.locale)

            self.formatter = logging.Formatter(self._config.log_format_str, datetime_fmt, style='{')

        else:

            self.formatter = None

    def _configure_stdout_handler(self):

        if self._config.log_to_stdout:

            config = self._config

            class StdErrBasedMaxLevelFilter(logging.Filter):

                def filter(self, record):

                    return record.levelno < config.stderr_log_min_level

            self.stdout_handler = StreamHandler(stdout)

            self.stdout_handler.setLevel(self._config.stdout_log_min_level)

            if self._config.log_to_stderr:

                self.stdout_handler.addFilter(StdErrBasedMaxLevelFilter())

        else:

            self.stdout_handler = None

    def _configure_stderr_handler(self) -> StreamHandler | None:

        if self._config.log_to_stderr:

            self.stderr_handler = StreamHandler(stderr)

            self.stderr_handler.setLevel(self._config.stderr_log_min_level)

        else:

            self.stderr_handler = None

    def _configure_file_handler(self) -> TimedRotatingFileHandler | None:

        if self._config.log_to_file:

            log_dir_path = self._config.file_log_dir_path

            if not os.path.exists(log_dir_path):

                os.makedirs(log_dir_path)

            log_file_path = Path(log_dir_path).joinpath('omnipy.log')

            self.file_handler = TimedRotatingFileHandler(

                log_file_path, when='d', interval=1, backupCount=7)

            self.file_handler.setLevel(self._config.file_log_min_level)

        else:

            self.file_handler = None

    def _remove_all_handlers_from_root_logger(self):

        root_logger = logging.root

        handlersToRemove = [

            handler for handler in root_logger.handlers

            if isinstance(handler, StreamHandler) or isinstance(handler, TimedRotatingFileHandler)

        ]

        for handler in handlersToRemove:

            root_logger.removeHandler(handler)

    def _add_handler_to_root_logger(self, handler: logging.Handler | None):

        if handler:

            root_logger = logging.root

            if self.formatter:

                handler.setFormatter(self.formatter)

            root_logger.addHandler(handler)

    def _add_all_handlers_to_root_logger(self):

        self._add_handler_to_root_logger(self.stdout_handler)

        self._add_handler_to_root_logger(self.stderr_handler)

        self._add_handler_to_root_logger(self.file_handler)

Class variables

file_handler

formatter

stderr_handler

stdout_handler

Methods

eq

def __eq__(
    self,
    other
)

Return self==value.

Parameters:

Name	Type	Description	Default
`other`

set_config

def set_config(
    self,
    config: omnipy.api.protocols.public.config.IsRootLogConfig
)

Parameters:

Name	Type	Description	Default
`config`	`IsRootLogConfig`

View Source

    def set_config(self, config: IsRootLogConfig):

        self._config = config

        self._configure_all_objects()