diff --git a/date_and_time/dt_formatting_styles.py b/date_and_time/dt_formatting_styles.py new file mode 100755 index 0000000..0b12b3f --- /dev/null +++ b/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}" diff --git a/logging_extensions/__init__.py b/logging_extensions/__init__.py new file mode 100755 index 0000000..e69de29 diff --git a/logging_extensions/handlers/__init__.py b/logging_extensions/handlers/__init__.py new file mode 100755 index 0000000..e69de29 diff --git a/logging_extensions/handlers/base_log_handler.py b/logging_extensions/handlers/base_log_handler.py new file mode 100755 index 0000000..4109a61 --- /dev/null +++ b/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()) diff --git a/logging_extensions/handlers/text_file_write_handler.py b/logging_extensions/handlers/text_file_write_handler.py new file mode 100755 index 0000000..9b2defd --- /dev/null +++ b/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 diff --git a/logging_extensions/log_message.py b/logging_extensions/log_message.py new file mode 100755 index 0000000..6c10446 --- /dev/null +++ b/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) diff --git a/logging_extensions/logger.py b/logging_extensions/logger.py new file mode 100755 index 0000000..d41ba9e --- /dev/null +++ b/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 diff --git a/logging_extensions/logging_config.py b/logging_extensions/logging_config.py new file mode 100755 index 0000000..036c822 --- /dev/null +++ b/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 [], + ) diff --git a/logging_extensions/severity/__init__.py b/logging_extensions/severity/__init__.py new file mode 100755 index 0000000..e69de29 diff --git a/logging_extensions/severity/log_levels.py b/logging_extensions/severity/log_levels.py new file mode 100755 index 0000000..df09e27 --- /dev/null +++ b/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) diff --git a/logging_extensions/severity/mapping.py b/logging_extensions/severity/mapping.py new file mode 100755 index 0000000..34ecd36 --- /dev/null +++ b/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 diff --git a/logging_extensions/severity/severity.py b/logging_extensions/severity/severity.py new file mode 100755 index 0000000..aa52ef0 --- /dev/null +++ b/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 diff --git a/macros/string_macros.py b/macros/string_macros.py index 6f4d32b..499fd63 100755 --- a/macros/string_macros.py +++ b/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)