Logger implementation

date_and_time/dt_formatting_styles.py

@@ -0,0 +1,35 @@
+from meta.config_meta import FinalConfigMeta
+from types_extensions import const, void
+
+
+class DateFormatting(metaclass=FinalConfigMeta):
+    DDMMYYYY: const(str) = '%d%m%Y'
+    YYYYMMDD: const(str) = '%Y%m%d'
+    DDMMYYYY_dash_delimited: const(str) = '%d-%m-%Y'
+    YYYYMMDD_dash_delimited: const(str) = '%Y-%m-%d'
+    DDMMYYYY_space_delimited: const(str) = '%d %m %Y'
+    YYYYMMDD_space_delimited: const(str) = '%Y %m %d'
+
+
+class TimeFormatting(metaclass=FinalConfigMeta):
+    HHMMSS: const(str) = '%H%M%S'
+    HHMMSS_dash_delimited: const(str) = '%H-%M-%S'
+    HHMMSS_space_delimited: const(str) = '%H %M %S'
+    HHMMSSms: const(str) = '%H%M%S%f'
+    HHMMSSms_dash_delimited: const(str) = '%H-%M-%S-%f'
+    HHMMSSms_space_delimited: const(str) = '%H %M %S %f'
+
+
+class DTFormatter(DateFormatting, TimeFormatting):
+
+    DT_DELIMITER: str = ' '
+
+    def __init__(self, delimiter: str = ' ') -> void:
+        if delimiter:
+            self.DT_DELIMITER = delimiter
+
+    def default_time_first(self) -> str:
+        return f"{self.HHMMSS_dash_delimited}{self.DT_DELIMITER}{self.DDMMYYYY_dash_delimited}"
+
+    def default_date_indexed(self) -> str:
+        return f"{self.YYYYMMDD_dash_delimited}{self.DT_DELIMITER}{self.HHMMSS_dash_delimited}"

logging_extensions/handlers/base_log_handler.py

@@ -0,0 +1,33 @@
+import abc
+
+from logging_extensions.log_message import LogMessage
+from logging_extensions.logging_config import LoggingConfig
+from types_extensions import void, const, safe_type
+
+
+class BaseLogHandler(metaclass=abc.ABCMeta):
+
+    def __init__(self, logger_name: str, parent_config: LoggingConfig, enabled: bool) -> void:
+        self.logger_name: const(str) = logger_name
+        self.config: LoggingConfig = parent_config
+        self.enabled: bool = enabled
+        self.current_log: safe_type(LogMessage) = None
+
+    @abc.abstractmethod
+    def handle_message(self, message: LogMessage, **kwargs) -> void:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def flush(self) -> void:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def enable(self) -> void:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def disable(self) -> void:
+        raise NotImplementedError
+
+    def _format_time_date_current(self):
+        return self.current_log.timestamp.strftime(self.config.DT_FORMATTER.default_time_first())

logging_extensions/handlers/text_file_write_handler.py

@@ -0,0 +1,54 @@
+import json
+from typing import Any
+
+from logging_extensions.handlers.base_log_handler import BaseLogHandler
+from logging_extensions.log_message import LogMessage
+from logging_extensions.logging_config import LoggingConfig
+from macros.string_macros import multiple_replace
+from types_extensions import void, dict_type
+
+
+class TextFileWriteHandler(BaseLogHandler):
+
+    def __init__(self, logger_name: str, parent_config: LoggingConfig, enabled: bool,
+                 text_log_file_location: str, **_) -> void:
+        super().__init__(logger_name, parent_config, enabled)
+        self.destination: str = text_log_file_location  # TODO: When safe pathing is merged, create safe path
+        self.text_buffer: str = ''
+
+    def handle_message(self, message: LogMessage, **kwargs) -> void:
+        if self.enabled and message.severity >= self.config.log_level:
+            self.current_log = message
+            self.text_buffer = self._format_current_as_string()
+        self.flush()
+
+    def flush(self) -> void:
+        if self.enabled:
+            self._flush()
+            self.text_buffer = ''
+            self.current_log = None
+
+    def _flush(self):
+        with open(self.destination, mode='a+') as log_file_h:
+            log_file_h.write(self.text_buffer)
+
+    def _format_current_as_string(self) -> str:
+        return multiple_replace(
+            self.config.string_log_fmt,
+            self._build_message_format_kwargs()
+        ) + "\n"
+
+    def _build_message_format_kwargs(self) -> dict_type[str: Any]:
+        return {
+            self.config.DT_IDENTIFIER: self._format_time_date_current(),
+            self.config.SEVERITY_IDENTIFIER: self.current_log.severity,
+            self.config.LOGGER_NAME_IDENTIFIER: self.logger_name,
+            self.config.MSG_IDENTIFIER: self.current_log.message,
+            self.config.FIELDS_IDENTIFIER: json.dumps(self.current_log.fields)
+        }
+
+    def enable(self) -> void:
+        self.enabled = True
+
+    def disable(self) -> void:
+        self.enabled = False

logging_extensions/log_message.py

@@ -0,0 +1,38 @@
+import dataclasses
+from datetime import datetime as _dt
+from typing import Any
+
+from logging_extensions.severity.log_levels import LogLevels, Severity
+from types_extensions import const, dict_type, void, list_type
+
+
+@dataclasses.dataclass(init=False)
+class LogMessage:
+
+    severity: const(Severity)
+    timestamp: const(_dt)
+    message: str
+    fields: dict_type[str, Any]
+
+    def __init__(self, message: str, timestamp: _dt, severity: Severity = LogLevels.INFO,
+                 fields: dict_type[str, Any] = None, message_format_mapping: dict_type[str, Any] = None):
+        self.timestamp: const(_dt) = timestamp
+        self.severity: const(Severity) = severity
+        self.fields: dict_type[str, Any] = fields
+        message_format_mapping = message_format_mapping or {}
+        self.message = message.format(**message_format_mapping)
+
+    def register_field(self, location: list_type[str], value: Any) -> void:
+        path = {}
+        curr = path
+        for i, loc_ in enumerate(location):
+            if i == len(location) - 1:
+                curr[loc_] = value
+            else:
+                curr[loc_] = {}
+                curr = curr[loc_]
+
+        self.fields.update()
+
+    def register_fields(self, fields: dict[str: Any]) -> void:
+        self.fields.update(fields)

logging_extensions/logger.py

@@ -0,0 +1,102 @@
+from typing import Any
+from datetime import datetime as _dt
+
+from logging_extensions.handlers.text_file_write_handler import TextFileWriteHandler
+from logging_extensions.severity.log_levels import LogLevels
+from logging_extensions.log_message import LogMessage
+from logging_extensions.logging_config import LoggingConfig
+from logging_extensions.severity.severity import Severity
+from types_extensions import void, const, dict_type, list_type
+from logging_extensions.handlers.base_log_handler import BaseLogHandler
+
+
+# zipped_b = gzip.compress(bytes("string", 'utf-8'), compresslevel=9)
+
+
+class LoggerPlus:
+
+    _DEFAULT_HANDLERS: list_type[BaseLogHandler] = [TextFileWriteHandler]
+
+    def __init__(self, name: str = None, config: LoggingConfig = None, enabled: bool = True,
+                 include_default_handlers: bool = True, **config_kwargs) -> void:
+        self._initialized: bool = False
+        self.name: const(str) = name or 'Unnamed'
+        self.enabled: bool = enabled
+        self.config: LoggingConfig = config or LoggingConfig.get_config(**config_kwargs)
+        self.handlers: list[BaseLogHandler] = []
+        self._init(include_default_handlers, **config_kwargs)
+
+    def _init(self, include_default_handlers: bool, **kwargs) -> void:
+        kwargs = kwargs or {}
+        handler_classes = self.config.handler_classes
+        if include_default_handlers:
+            handler_classes += self._DEFAULT_HANDLERS
+        for handler_class in self.config.handler_classes:
+            handler = handler_class(
+                logger_name=self.name,
+                parent_config=self.config,
+                enabled=self.enabled,
+                **kwargs
+            )
+            self.handlers.append(handler)
+        self._initialized = True
+
+    def _log(self, severity: Severity, message: str = '', fields: dict_type[str, Any] = None,
+             message_format_mapping: dict_type[str, Any] = None, **handler_kwargs) -> void:
+        message_ = LogMessage(message=message,
+                              timestamp=_dt.now(),
+                              severity=severity,
+                              fields=fields,
+                              message_format_mapping=message_format_mapping)
+        for handler in self.handlers:
+            handler.handle_message(message=message_, **handler_kwargs)
+
+    def debug(self, message: str = '', fields: dict_type[str, Any] = None,
+              message_format_mapping: dict_type[str, Any] = None, **handler_kwargs):
+        self._log(
+            LogLevels.DEBUG,
+            message,
+            fields,
+            message_format_mapping,
+            **handler_kwargs
+        )
+
+    def info(self, message: str = '', fields: dict_type[str, Any] = None,
+             message_format_mapping: dict_type[str, Any] = None, **handler_kwargs):
+        self._log(
+            LogLevels.INFO,
+            message,
+            fields,
+            message_format_mapping,
+            **handler_kwargs
+        )
+
+    def warning(self, message: str = '', fields: dict_type[str, Any] = None,
+                message_format_mapping: dict_type[str, Any] = None, **handler_kwargs):
+        self._log(
+            LogLevels.WARNING,
+            message,
+            fields,
+            message_format_mapping,
+            **handler_kwargs
+        )
+
+    def error(self, message: str = '', fields: dict_type[str, Any] = None,
+              message_format_mapping: dict_type[str, Any] = None, **handler_kwargs):
+        self._log(
+            LogLevels.ERROR,
+            message,
+            fields,
+            message_format_mapping,
+            **handler_kwargs
+        )
+
+    def enable(self) -> void:
+        self.enabled = True
+        for handler in self.handlers:
+            handler.enabled = True
+
+    def disable(self) -> void:
+        self.enabled = False
+        for handler in self.handlers:
+            handler.enabled = False

logging_extensions/logging_config.py

@@ -0,0 +1,32 @@
+from date_and_time.dt_formatting_styles import DTFormatter
+from logging_extensions.severity.log_levels import LogLevels, Severity
+from meta.config_meta import BaseConfig
+from types_extensions import list_type, const, void
+
+
+class LoggingConfig(BaseConfig):
+    log_level: Severity
+    compression_level: int
+    handler_classes: list_type[type]
+    DT_FORMATTER: const(DTFormatter) = DTFormatter()
+    DT_IDENTIFIER: const(str) = '$$datetime'
+    SEVERITY_IDENTIFIER: const(str) = '$$severity'
+    LOGGER_NAME_IDENTIFIER: const(str) = '$$name'
+    MSG_IDENTIFIER: const(str) = '$$msg'
+    FIELDS_IDENTIFIER: const(str) = '$$fields'
+    string_log_fmt: str = f"[{DT_IDENTIFIER}][{SEVERITY_IDENTIFIER}][{LOGGER_NAME_IDENTIFIER}] " \
+                          f"<<{MSG_IDENTIFIER}>> FIELDS: {FIELDS_IDENTIFIER}"
+
+    def __init__(self, log_level: Severity, compression_level: int, handler_classes: list_type[type]) -> void:
+        self.log_level: Severity = log_level
+        self.compression_level: int = compression_level
+        self.handler_classes: list_type[type] = handler_classes
+
+    @classmethod
+    def get_config(cls, *, log_level: str = None, compression_level: int = 9,
+                   handler_classes: list_type[type] = None, **__) -> 'LoggingConfig':
+        return LoggingConfig(
+            log_level=log_level or LogLevels.INFO,
+            compression_level=compression_level,
+            handler_classes=handler_classes or [],
+        )

logging_extensions/severity/log_levels.py

@@ -0,0 +1,12 @@
+from logging_extensions.severity.mapping import SeverityLabelMapping, SeverityLevelMapping
+from logging_extensions.severity.severity import Severity
+from meta.config_meta import FinalConfigMeta
+from types_extensions import const
+
+
+class LogLevels(metaclass=FinalConfigMeta):
+
+    DEBUG: const(Severity) = Severity(text=SeverityLabelMapping.DEBUG, level=SeverityLevelMapping.DEBUG)
+    INFO: const(Severity) = Severity(text=SeverityLabelMapping.INFO, level=SeverityLevelMapping.INFO)
+    WARNING: const(Severity) = Severity(text=SeverityLabelMapping.WARNING, level=SeverityLevelMapping.WARNING)
+    ERROR: const(Severity) = Severity(text=SeverityLabelMapping.ERROR, level=SeverityLevelMapping.ERROR)

logging_extensions/severity/mapping.py

@@ -0,0 +1,16 @@
+from meta.config_meta import FinalConfigMeta
+from types_extensions import const
+
+
+class SeverityLabelMapping(FinalConfigMeta):
+    DEBUG: const(str) = "DEBUG"
+    INFO: const(str) = "INFO"
+    WARNING: const(str) = "WARNING"
+    ERROR: const(str) = "ERROR"
+
+
+class SeverityLevelMapping(FinalConfigMeta):
+    DEBUG: const(int) = 0
+    INFO: const(int) = 10
+    WARNING: const(int) = 70
+    ERROR: const(int) = 95

logging_extensions/severity/severity.py

@@ -0,0 +1,24 @@
+import dataclasses
+
+from types_extensions import const
+
+
+@dataclasses.dataclass
+class Severity:
+    text: const(str)
+    level: const(int)
+
+    def __str__(self) -> str:
+        return self.text
+
+    def __repr__(self) -> str:
+        return str(self)
+
+    def __eq__(self, other: 'Severity'):
+        return self.level == other.level
+
+    def __gt__(self, other: 'Severity'):
+        return self.level > other.level
+
+    def __ge__(self, other: 'Severity'):
+        return self.level >= other.level

macros/string_macros.py

@@ -1,3 +1,5 @@
+import re
+from typing import Any
 from types_extensions import tuple_type
 
 
@@ -8,3 +10,12 @@ def split_string(str_: str, index: int) -> tuple_type[str, str]:
     if index >= len(str_):
         raise IndexError
     return str_[:index], str_[index:]
+
+
+def multiple_replace(input_string: str, kwargs_dict: dict[str: Any]):
+    """
+    Replaces each occurring key (from kwargs_dict) with its value in input_string.
+    Useful in cases where f-strings or {} formatting can't be used
+    """
+    pattern = re.compile("|".join([re.escape(key) for key in kwargs_dict.keys()]))
+    return pattern.sub(lambda x: str(kwargs_dict[x.group(0)]), input_string)