sh-edraft.de/cpl
5
0
Fork 0
Code Issues 5 Pull Requests 2 Actions Packages Projects 1 Releases 12 Wiki Activity

Improved logging closes #180
All checks were successful
Build on push / prepare (push) Successful in 9s
Details
Build on push / query (push) Successful in 17s
Details
Build on push / core (push) Successful in 18s
Details
Build on push / mail (push) Successful in 17s
Details
Build on push / translation (push) Successful in 18s
Details

Browse Source
This commit is contained in:
Sven Heidemann
2025-09-15 23:06:38 +02:00
parent d9d7802f33
commit 62b6435470
35 changed files with 533 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/cpl-core/cpl/core/application/application_builder.py
View File

@@ -3,6 +3,9 @@ from typing import Type, Optional, Callable, Union
from cpl.core.application.application_abc import ApplicationABC
from cpl.core.application.application_builder_abc import ApplicationBuilderABC
from cpl.core.application.application_extension_abc import ApplicationExtensionABC
from cpl.core.application.async_application_extension_abc import AsyncApplicationExtensionABC
from cpl.core.application.async_startup_abc import AsyncStartupABC
from cpl.core.application.async_startup_extension_abc import AsyncStartupExtensionABC
from cpl.core.application.startup_abc import StartupABC
from cpl.core.application.startup_extension_abc import StartupExtensionABC
from cpl.core.configuration.configuration import Configuration
@@ -20,25 +23,25 @@ class ApplicationBuilder(ApplicationBuilderABC):
def __init__(self, app: Type[ApplicationABC]):
ApplicationBuilderABC.__init__(self)
self._app = app
self._startup: Optional[StartupABC] = None
self._startup: Optional[StartupABC | AsyncStartupABC] = None
self._configuration = Configuration()
self._environment = self._configuration.environment
self._services = ServiceCollection(self._configuration)
self._app_extensions: list[Callable] = []
self._startup_extensions: list[Callable] = []
self._app_extensions: list[Type[ApplicationExtensionABC | AsyncApplicationExtensionABC]] = []
self._startup_extensions: list[Type[StartupExtensionABC | AsyncStartupABC]] = []
def use_startup(self, startup: Type[StartupABC]) -> "ApplicationBuilder":
def use_startup(self, startup: Type[StartupABC | AsyncStartupABC]) -> "ApplicationBuilder":
self._startup = startup()
return self
def use_extension(
self, extension: Type[Union[ApplicationExtensionABC, StartupExtensionABC]]
self, extension: Type[ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC]
) -> "ApplicationBuilder":
if issubclass(extension, ApplicationExtensionABC) and extension not in self._app_extensions:
if (issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)) and extension not in self._app_extensions:
self._app_extensions.append(extension)
elif issubclass(extension, StartupExtensionABC) and extension not in self._startup_extensions:
elif (issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)) and extension not in self._startup_extensions:
self._startup_extensions.append(extension)
return self
@@ -53,6 +56,16 @@ class ApplicationBuilder(ApplicationBuilderABC):
self._startup.configure_configuration(self._configuration, self._environment)
self._startup.configure_services(self._services, self._environment)
async def _build_async_startup(self):
for ex in self._startup_extensions:
extension = ex()
await extension.configure_configuration(self._configuration, self._environment)
await extension.configure_services(self._services, self._environment)
if self._startup is not None:
await self._startup.configure_configuration(self._configuration, self._environment)
await self._startup.configure_services(self._services, self._environment)
def build(self) -> ApplicationABC:
self._build_startup()
@@ -66,7 +79,7 @@ class ApplicationBuilder(ApplicationBuilderABC):
return self._app(config, services)
async def build_async(self) -> ApplicationABC:
self._build_startup()
await self._build_async_startup()
config = self._configuration
services = self._services.build_service_provider()

4
src/cpl-core/cpl/core/application/application_extension_abc.py
View File

@@ -12,7 +12,3 @@ class ApplicationExtensionABC(ABC):
@abstractmethod
def run(self, config: ConfigurationABC, services: ServiceProviderABC):
pass
@abstractmethod
async def run(self, config: ConfigurationABC, services: ServiceProviderABC):
pass

14
src/cpl-core/cpl/core/application/async_application_extension_abc.py Normal file
View File

@@ -0,0 +1,14 @@
from abc import ABC, abstractmethod
from cpl.core.configuration import ConfigurationABC
from cpl.core.dependency_injection import ServiceProviderABC
class AsyncApplicationExtensionABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def run(self, config: ConfigurationABC, services: ServiceProviderABC):
pass

32
src/cpl-core/cpl/core/application/async_startup_abc.py Normal file
View File

@@ -0,0 +1,32 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
class AsyncStartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
async def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

31
src/cpl-core/cpl/core/application/async_startup_extension_abc.py Normal file
View File

@@ -0,0 +1,31 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
class AsyncStartupExtensionABC(ABC):
r"""ABC for startup extension classes"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
async def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

10
src/cpl-core/cpl/core/application/startup_abc.py
View File

@@ -14,25 +14,19 @@ class StartupABC(ABC):
pass
@abstractmethod
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC) -> ConfigurationABC:
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl.core.environment.application_environment_abc`
Returns:
Object of :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
"""
@abstractmethod
def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC) -> ServiceProviderABC:
def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
Returns:
Object of :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
"""

47
src/cpl-core/cpl/core/configuration/configuration.py
View File

@@ -22,7 +22,7 @@ from cpl.core.dependency_injection.service_provider_abc import ServiceProviderAB
from cpl.core.environment.application_environment import ApplicationEnvironment
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.environment.environment_name_enum import EnvironmentNameEnum
from cpl.core.type import T, R
from cpl.core.typing import T, R
from cpl.core.utils.json_processor import JSONProcessor
@@ -62,7 +62,7 @@ class Configuration(ConfigurationABC):
return self._argument_types
@staticmethod
def _print_info(name: str, message: str):
def _print_info(message: str):
r"""Prints an info message
Parameter:
@@ -72,11 +72,11 @@ class Configuration(ConfigurationABC):
Info message
"""
Console.set_foreground_color(ForegroundColorEnum.green)
Console.write_line(f"[{name}] {message}")
Console.write_line(f"[CONF] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
@staticmethod
def _print_warn(name: str, message: str):
def _print_warn(message: str):
r"""Prints a warning
Parameter:
@@ -86,11 +86,11 @@ class Configuration(ConfigurationABC):
Warning message
"""
Console.set_foreground_color(ForegroundColorEnum.yellow)
Console.write_line(f"[{name}] {message}")
Console.write_line(f"[CONF] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
@staticmethod
def _print_error(name: str, message: str):
def _print_error(message: str):
r"""Prints an error
Parameter:
@@ -100,7 +100,7 @@ class Configuration(ConfigurationABC):
Error message
"""
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f"[{name}] {message}")
Console.write_line(f"[CONF] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
def _set_variable(self, name: str, value: any):
@@ -142,40 +142,13 @@ class Configuration(ConfigurationABC):
# load json
json_cfg = json.load(cfg)
if output:
self._print_info(__name__, f"Loaded config file: {file}")
self._print_info(f"Loaded config file: {file}")
return json_cfg
except Exception as e:
self._print_error(__name__, f"Cannot load config file: {file}! -> {e}")
self._print_error(f"Cannot load config file: {file}! -> {e}")
return {}
def _handle_pre_or_post_executables(self, pre: bool, argument: ExecutableArgument, services: ServiceProviderABC):
script_type = "pre-" if pre else "post-"
from cpl_cli.configuration.workspace_settings import WorkspaceSettings
workspace: Optional[WorkspaceSettings] = self.get_configuration(WorkspaceSettings)
if workspace is None or len(workspace.scripts) == 0:
return
for script in workspace.scripts:
if script_type not in script and not script.startswith(script_type):
continue
# split in two ifs to prevent exception
if script.split(script_type)[1] != argument.name:
continue
from cpl_cli.command.custom_script_service import CustomScriptService
css: CustomScriptService = services.get_service(CustomScriptService)
if css is None:
continue
Console.write_line()
self._set_variable("ACTIVE_EXECUTABLE", script)
css.run(self._additional_arguments)
def _parse_arguments(
self,
executables: list[ArgumentABC],
@@ -264,7 +237,7 @@ class Configuration(ConfigurationABC):
if not os.path.isfile(file_path):
if optional is not True:
if output:
self._print_error(__name__, f"File not found: {file_path}")
self._print_error(f"File not found: {file_path}")
sys.exit()

2
src/cpl-core/cpl/core/configuration/configuration_abc.py
View File

@@ -5,7 +5,7 @@ from typing import Optional
from cpl.core.configuration.argument_abc import ArgumentABC
from cpl.core.configuration.argument_type_enum import ArgumentTypeEnum
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.type import T, R
from cpl.core.typing import T, R
class ConfigurationABC(ABC):

8
src/cpl-core/cpl/core/database/db_logger.py Normal file
View File

@@ -0,0 +1,8 @@
from cpl.core.log import Logger
from cpl.core.typing import Source
class DBLogger(Logger):
def __init__(self, source: Source):
Logger.__init__(self, source, "db")

12
src/cpl-core/cpl/core/dependency_injection/service_collection.py
View File

@@ -8,10 +8,10 @@ from cpl.core.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.core.dependency_injection.service_provider import ServiceProvider
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.log.logger import Logger
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.log.logger_service import Logger
from cpl.core.pipes.pipe_abc import PipeABC
from cpl.core.type import T
from cpl.core.typing import T, Service
class ServiceCollection(ServiceCollectionABC):
@@ -53,7 +53,7 @@ class ServiceCollection(ServiceCollectionABC):
self._database_context.connect(db_settings)
def add_logging(self):
self.add_singleton(LoggerABC, Logger)
self.add_transient(LoggerABC, Logger)
return self
def add_pipes(self):
@@ -61,15 +61,15 @@ class ServiceCollection(ServiceCollectionABC):
self.add_transient(PipeABC, pipe)
return self
def add_singleton(self, service_type: T, service: T = None):
def add_singleton(self, service_type: T, service: Service = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.singleton, service)
return self
def add_scoped(self, service_type: T, service: T = None):
def add_scoped(self, service_type: T, service: Service = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.scoped, service)
return self
def add_transient(self, service_type: T, service: T = None):
def add_transient(self, service_type: T, service: Service = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.transient, service)
return self

12
src/cpl-core/cpl/core/dependency_injection/service_collection_abc.py
View File

@@ -1,10 +1,10 @@
from abc import abstractmethod, ABC
from typing import Type
from cpl.core.database.database_settings import DatabaseSettings
from cpl.core.database.context.database_context_abc import DatabaseContextABC
from cpl.core.database.database_settings import DatabaseSettings
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.type import T
from cpl.core.typing import T, Source
class ServiceCollectionABC(ABC):
@@ -31,14 +31,14 @@ class ServiceCollectionABC(ABC):
def add_pipes(self):
r"""Adds the CPL internal pipes as transient"""
def add_discord(self):
r"""Adds the CPL discord"""
raise NotImplementedError("You should install and use the cpl-discord package")
def add_translation(self):
r"""Adds the CPL translation"""
raise NotImplementedError("You should install and use the cpl-translation package")
def add_mail(self):
r"""Adds the CPL mail"""
raise NotImplementedError("You should install and use the cpl-mail package")
@abstractmethod
def add_transient(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with transient lifetime

29
src/cpl-core/cpl/core/dependency_injection/service_provider.py
View File

@@ -12,7 +12,7 @@ from cpl.core.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.type import T, R
from cpl.core.typing import T, R, Source
class ServiceProvider(ServiceProviderABC):
@@ -48,7 +48,7 @@ class ServiceProvider(ServiceProviderABC):
return None
def _get_service(self, parameter: Parameter) -> Optional[object]:
def _get_service(self, parameter: Parameter, origin_service_type: type = None) -> Optional[object]:
for descriptor in self._service_descriptors:
if descriptor.service_type == parameter.annotation or issubclass(
descriptor.service_type, parameter.annotation
@@ -56,7 +56,7 @@ class ServiceProvider(ServiceProviderABC):
if descriptor.implementation is not None:
return descriptor.implementation
implementation = self.build_service(descriptor.service_type)
implementation = self._build_service(descriptor.service_type, origin_service_type=origin_service_type)
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptor.implementation = implementation
@@ -64,7 +64,7 @@ class ServiceProvider(ServiceProviderABC):
# raise Exception(f'Service {parameter.annotation} not found')
def _get_services(self, t: type, *args, **kwargs) -> list[Optional[object]]:
def _get_services(self, t: type, *args, service_type: type = None, **kwargs) -> list[Optional[object]]:
implementations = []
for descriptor in self._service_descriptors:
if descriptor.service_type == t or issubclass(descriptor.service_type, t):
@@ -72,7 +72,7 @@ class ServiceProvider(ServiceProviderABC):
implementations.append(descriptor.implementation)
continue
implementation = self.build_service(descriptor.service_type, *args, **kwargs)
implementation = self._build_service(descriptor.service_type, *args, service_type, **kwargs)
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptor.implementation = implementation
@@ -80,13 +80,16 @@ class ServiceProvider(ServiceProviderABC):
return implementations
def build_by_signature(self, sig: Signature) -> list[R]:
def _build_by_signature(self, sig: Signature, origin_service_type: type) -> list[R]:
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != "self" and parameter.annotation != Parameter.empty:
if typing.get_origin(parameter.annotation) == list:
params.append(self._get_services(typing.get_args(parameter.annotation)[0]))
params.append(self._get_services(typing.get_args(parameter.annotation)[0], origin_service_type))
elif parameter.annotation == Source:
params.append(origin_service_type.__name__)
elif issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
@@ -104,11 +107,15 @@ class ServiceProvider(ServiceProviderABC):
params.append(self._configuration)
else:
params.append(self._get_service(parameter))
params.append(self._get_service(parameter, origin_service_type))
return params
def build_service(self, service_type: type, *args, **kwargs) -> object:
def _build_service(self, service_type: type, *args, origin_service_type: type = None, **kwargs) -> object:
if origin_service_type is None:
origin_service_type = service_type
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
if descriptor.implementation is not None:
@@ -119,7 +126,7 @@ class ServiceProvider(ServiceProviderABC):
break
sig = signature(service_type.__init__)
params = self.build_by_signature(sig)
params = self._build_by_signature(sig, origin_service_type)
return service_type(*params, *args, **kwargs)
@@ -147,7 +154,7 @@ class ServiceProvider(ServiceProviderABC):
if result.implementation is not None:
return result.implementation
implementation = self.build_service(service_type, *args, **kwargs)
implementation = self._build_service(service_type, *args, **kwargs)
if (
result.lifetime == ServiceLifetimeEnum.singleton
or result.lifetime == ServiceLifetimeEnum.scoped

8
src/cpl-core/cpl/core/dependency_injection/service_provider_abc.py
View File

@@ -4,7 +4,7 @@ from inspect import Signature, signature
from typing import Optional
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.core.type import T, R
from cpl.core.typing import T, R
class ServiceProviderABC(ABC):
@@ -21,11 +21,11 @@ class ServiceProviderABC(ABC):
cls._provider = provider
@abstractmethod
def build_by_signature(self, sig: Signature) -> list[R]:
def _build_by_signature(self, sig: Signature, origin_service_type: type) -> list[R]:
pass
@abstractmethod
def build_service(self, service_type: type, *args, **kwargs) -> object:
def _build_service(self, service_type: type, *args, **kwargs) -> object:
r"""Creates instance of given type
Parameter
@@ -105,7 +105,7 @@ class ServiceProviderABC(ABC):
if cls._provider is None:
raise Exception(f"{cls.__name__} not build!")
injection = [x for x in cls._provider.build_by_signature(signature(f)) if x is not None]
injection = [x for x in cls._provider._build_by_signature(signature(f)) if x is not None]
return f(*args, *injection, **kwargs)
return inner

32
src/cpl-core/cpl/core/environment/environment.py Normal file
View File

@@ -0,0 +1,32 @@
import os
from typing import Optional, Type
from cpl.core.typing import T
from cpl.core.utils.get_value import get_value
class Environment:
_environment = "production"
@classmethod
def get_environment(cls):
return cls._environment
@classmethod
def set_environment(cls, environment: str):
if environment not in ["development", "staging", "production"]:
raise ValueError("Invalid environment")
Environment._environment = environment
@staticmethod
def get(key: str, cast_type: Type[T], default: Optional[T] = None) -> Optional[T]:
"""
Get an environment variable and cast it to a specified type.
:param str key: The name of the environment variable.
:param Type[T] cast_type: A callable to cast the variable's value.
:param Optional[T] default: The default value to return if the variable is not found. Defaults to None.The default value to return if the variable is not found. Defaults to None.
:return: The casted value, or None if the variable is not found.
:rtype: Optional[T]
"""
return get_value(dict(os.environ), key, cast_type, default)

7
src/cpl-core/cpl/core/log/__init__.py
View File

@@ -1,5 +1,4 @@
from .logger_service import Logger
from .logger import Logger
from .logger_abc import LoggerABC
from .logging_level_enum import LoggingLevelEnum
from .logging_settings import LoggingSettings
from .logging_settings_name_enum import LoggingSettingsNameEnum
from .log_level_enum import LogLevelEnum
from .logging_settings import LogSettings

94
src/cpl-core/cpl/core/log/_log_writer.py Normal file
View File

@@ -0,0 +1,94 @@
import multiprocessing
import os
from datetime import datetime
from typing import Self
from cpl.core.console import Console
from cpl.core.log.log_level_enum import LogLevelEnum
class LogWriter:
_instance = None
# ANSI color codes for different log levels
_COLORS = {
LogLevelEnum.trace: "\033[37m", # Light Gray
LogLevelEnum.debug: "\033[94m", # Blue
LogLevelEnum.info: "\033[92m", # Green
LogLevelEnum.warning: "\033[93m", # Yellow
LogLevelEnum.error: "\033[91m", # Red
LogLevelEnum.fatal: "\033[95m", # Magenta
}
def __init__(self, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info):
self._file_prefix = file_prefix
self._level = level
self._queue = multiprocessing.Queue()
self._process = multiprocessing.Process(target=self._log_worker, daemon=True)
self._create_log_dir()
self._process.start()
@property
def level(self) -> LogLevelEnum:
return self._level
@level.setter
def level(self, value: LogLevelEnum):
assert isinstance(value, LogLevelEnum), "Log level must be an instance of LogLevelEnum"
self._level = value
@classmethod
def get_instance(cls, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info) -> Self:
if cls._instance is None:
cls._instance = LogWriter(file_prefix, level)
return cls._instance
@staticmethod
def _create_log_dir():
if os.path.exists("logs"):
return
os.makedirs("logs")
def _log_worker(self):
"""Worker process that writes log messages from the queue to the file."""
while True:
content = self._queue.get()
if content is None: # Shutdown signal
break
self._write_log_to_file(content)
Console.write_line(
f"{self._COLORS.get(self._level, '\033[0m')}{content}\033[0m"
)
@property
def log_file(self):
return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
def _ensure_file_size(self):
log_file = self.log_file
if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
return
# if exists and size is greater than 300MB, create a new file
os.rename(
log_file,
f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
)
def _write_log_to_file(self, content: str):
self._ensure_file_size()
with open(self.log_file, "a") as log_file:
log_file.write(content + "\n")
log_file.close()
def log(self, content: str):
"""Enqueue log message without blocking main app."""
self._queue.put(content)
def close(self):
"""Gracefully stop the logging process."""
self._queue.put(None)
self._process.join()

11
src/cpl-core/cpl/core/log/log_level_enum.py Normal file
View File

@@ -0,0 +1,11 @@
from enum import Enum
class LogLevelEnum(Enum):
off = "OFF" # Nothing
trace = "TRC" # Detailed app information's
debug = "DEB" # Detailed app state
info = "INF" # Normal information's
warning = "WAR" # Error that can later be fatal
error = "ERR" # Non fatal error
fatal = "FAT" # Error that cause exit

78
src/cpl-core/cpl/core/log/logger.py Normal file
View File

@@ -0,0 +1,78 @@
import os
import traceback
from datetime import datetime
from cpl.core.console import Console
from cpl.core.log._log_writer import LogWriter
from cpl.core.log.log_level_enum import LogLevelEnum
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.typing import Messages, Source
class Logger(LoggerABC):
_level = LogLevelEnum.info
_levels = [x for x in LogLevelEnum]
def __init__(self, source: Source, file_prefix: str = None):
LoggerABC.__init__(self)
assert source is not None and source != "", "Source cannot be None or empty"
self._source = source
if file_prefix is None:
file_prefix = "app"
self._file_prefix = file_prefix
self._writer = LogWriter.get_instance(self._file_prefix)
@classmethod
def set_level(cls, level: LogLevelEnum):
if level in cls._levels:
cls._level = level
else:
raise ValueError(f"Invalid log level: {level}")
def _log(self, level: LogLevelEnum, *messages: Messages):
try:
if self._levels.index(level) < self._levels.index(self._level):
return
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
formatted_message = self._format_message(level.value, timestamp, *messages)
self._writer.log(formatted_message)
except Exception as e:
print(f"Error while logging: {e} -> {traceback.format_exc()}")
def _format_message(self, level: str, timestamp, *messages: Messages) -> str:
if isinstance(messages, tuple):
messages = list(messages)
if not isinstance(messages, list):
messages = [messages]
messages = [str(message) for message in messages if message is not None]
return f"<{timestamp}> [{level.upper():^3}] [{self._file_prefix}] - [{self._source}]: {' '.join(messages)}"
def header(self, string: str):
self._log(LogLevelEnum.info, string)
def trace(self, *messages: Messages):
self._log(LogLevelEnum.trace, *messages)
def debug(self, *messages: Messages):
self._log(LogLevelEnum.debug, *messages)
def info(self, *messages: Messages):
self._log(LogLevelEnum.info, *messages)
def warning(self, *messages: Messages):
self._log(LogLevelEnum.warning, *messages)
def error(self, message, e: Exception = None):
self._log(LogLevelEnum.error, message, f"{e} -> {traceback.format_exc()}" if e else None)
def fatal(self, message, e: Exception = None, prevent_quit: bool = False):
self._log(LogLevelEnum.fatal, message, f"{e} -> {traceback.format_exc()}" if e else None)
if not prevent_quit:
exit(-1)

22
src/cpl-core/cpl/core/log/logger_abc.py
View File

@@ -1,12 +1,18 @@
from abc import abstractmethod, ABC
from cpl.core.typing import Messages
class LoggerABC(ABC):
r"""ABC for :class:`cpl.core.log.logger_service.Logger`"""
@abstractmethod
def __init__(self):
ABC.__init__(self)
def set_level(self, level: str):
pass
@abstractmethod
def _format_message(self, level: str, timestamp, *messages: Messages) -> str:
pass
@abstractmethod
def header(self, string: str):
@@ -18,7 +24,7 @@ class LoggerABC(ABC):
"""
@abstractmethod
def trace(self, name: str, message: str):
def trace(self, *messages: Messages):
r"""Writes a trace message
Parameter:
@@ -29,7 +35,7 @@ class LoggerABC(ABC):
"""
@abstractmethod
def debug(self, name: str, message: str):
def debug(self, *messages: Messages):
r"""Writes a debug message
Parameter:
@@ -40,7 +46,7 @@ class LoggerABC(ABC):
"""
@abstractmethod
def info(self, name: str, message: str):
def info(self, *messages: Messages):
r"""Writes an information
Parameter:
@@ -51,7 +57,7 @@ class LoggerABC(ABC):
"""
@abstractmethod
def warn(self, name: str, message: str):
def warning(self, *messages: Messages):
r"""Writes an warning
Parameter:
@@ -62,7 +68,7 @@ class LoggerABC(ABC):
"""
@abstractmethod
def error(self, name: str, message: str, ex: Exception = None):
def error(self, messages: str, e: Exception = None):
r"""Writes an error
Parameter:
@@ -75,7 +81,7 @@ class LoggerABC(ABC):
"""
@abstractmethod
def fatal(self, name: str, message: str, ex: Exception = None):
def fatal(self, messages: str, e: Exception = None):
r"""Writes an error and ends the program
Parameter:

291
src/cpl-core/cpl/core/log/logger_service.py
View File

@@ -1,291 +0,0 @@
import datetime
import os
import sys
import traceback
from string import Template
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.core.console.console import Console
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.log.logging_level_enum import LoggingLevelEnum
from cpl.core.log.logging_settings import LoggingSettings
from cpl.core.time.time_format_settings import TimeFormatSettings
class Logger(LoggerABC):
r"""Service for logging
Parameter:
logging_settings: :class:`cpl.core.log.logging_settings.LoggingSettings`
Settings for the logger
time_format: :class:`cpl.core.time.time_format_settings.TimeFormatSettings`
Time format settings
env: :class:`cpl.core.environment.application_environment_abc.ApplicationEnvironmentABC`
Environment of the application
"""
def __init__(
self, logging_settings: LoggingSettings, time_format: TimeFormatSettings, env: ApplicationEnvironmentABC
):
LoggerABC.__init__(self)
self._env = env
self._log_settings: LoggingSettings = logging_settings
self._time_format_settings: TimeFormatSettings = time_format
self._check_for_settings(self._time_format_settings, TimeFormatSettings)
self._check_for_settings(self._log_settings, LoggingSettings)
self._level = self._log_settings.level
self._console = self._log_settings.console
self.create()
@property
def _log(self) -> str:
return Template(self._log_settings.filename).substitute(
date_time_now=self._env.date_time_now.strftime(self._time_format_settings.date_time_format),
date_now=self._env.date_time_now.strftime(self._time_format_settings.date_format),
time_now=self._env.date_time_now.strftime(self._time_format_settings.time_format),
start_time=self._env.start_time.strftime(self._time_format_settings.date_time_log_format),
)
@property
def _path(self) -> str:
return Template(self._log_settings.path).substitute(
date_time_now=self._env.date_time_now.strftime(self._time_format_settings.date_time_format),
date_now=self._env.date_time_now.strftime(self._time_format_settings.date_format),
time_now=self._env.date_time_now.strftime(self._time_format_settings.time_format),
start_time=self._env.start_time.strftime(self._time_format_settings.date_time_log_format),
)
def _check_for_settings(self, settings: ConfigurationModelABC, settings_type: type):
self._level = LoggingLevelEnum.OFF
self._console = LoggingLevelEnum.FATAL
if settings is None:
self.fatal(__name__, f"Configuration for {settings_type} not found")
def _get_datetime_now(self) -> str:
r"""Returns the date and time by given format
Returns:
Date and time in given format
"""
try:
return datetime.datetime.now().strftime(self._time_format_settings.date_time_format)
except Exception as e:
self.error(__name__, "Cannot get time", ex=e)
def _get_date(self) -> str:
r"""Returns the date by given format
Returns:
Date in given format
"""
try:
return datetime.datetime.now().strftime(self._time_format_settings.date_format)
except Exception as e:
self.error(__name__, "Cannot get date", ex=e)
def create(self) -> None:
r"""Creates path tree and logfile"""
""" path """
try:
# check if log file path exists
if not os.path.exists(self._path):
os.makedirs(self._path)
except Exception as e:
self._fatal_console(__name__, "Cannot create log dir", ex=e)
""" create new log file """
try:
# open log file, create if not exists
path = f"{self._path}{self._log}"
permission = "a+"
if not os.path.isfile(path):
permission = "w+"
f = open(path, permission)
Console.write_line(f"[{__name__}]: Using log file: {path}")
f.close()
except Exception as e:
self._fatal_console(__name__, "Cannot open log file", ex=e)
def _append_log(self, string: str):
r"""Writes to logfile
Parameter:
string: :class:`str`
"""
try:
# open log file and append always
if not os.path.isdir(self._path):
self._warn_console(__name__, "Log directory not found, try to recreate logger")
self.create()
with open(self._path + self._log, "a+", encoding="utf-8") as f:
f.write(string + "\n")
f.close()
except Exception as e:
self._fatal_console(__name__, f"Cannot append log file, message: {string}", ex=e)
def _get_string(self, name: str, level: LoggingLevelEnum, message: str) -> str:
r"""Returns input as log entry format
Parameter:
name: :class:`str`
Name of the message
level: :class:`cpl.core.log.logging_level_enum.LoggingLevelEnum`
Logging level
message: :class:`str`
Log message
Returns:
Formatted string for logging
"""
log_level = level.name
return f"<{self._get_datetime_now()}> [ {log_level} ] [ {name} ]: {message}"
def header(self, string: str):
# append log and print message
self._append_log(string)
Console.set_foreground_color(ForegroundColorEnum.default)
Console.write_line(string)
Console.set_foreground_color(ForegroundColorEnum.default)
def trace(self, name: str, message: str):
output = self._get_string(name, LoggingLevelEnum.TRACE, message)
# check if message can be written to log
if self._level.value >= LoggingLevelEnum.TRACE.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevelEnum.TRACE.value:
Console.set_foreground_color(ForegroundColorEnum.grey)
Console.write_line(output)
Console.set_foreground_color(ForegroundColorEnum.default)
def debug(self, name: str, message: str):
output = self._get_string(name, LoggingLevelEnum.DEBUG, message)
# check if message can be written to log
if self._level.value >= LoggingLevelEnum.DEBUG.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevelEnum.DEBUG.value:
Console.set_foreground_color(ForegroundColorEnum.blue)
Console.write_line(output)
Console.set_foreground_color(ForegroundColorEnum.default)
def info(self, name: str, message: str):
output = self._get_string(name, LoggingLevelEnum.INFO, message)
# check if message can be written to log
if self._level.value >= LoggingLevelEnum.INFO.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevelEnum.INFO.value:
Console.set_foreground_color(ForegroundColorEnum.green)
Console.write_line(output)
Console.set_foreground_color(ForegroundColorEnum.default)
def warn(self, name: str, message: str):
output = self._get_string(name, LoggingLevelEnum.WARN, message)
# check if message can be written to log
if self._level.value >= LoggingLevelEnum.WARN.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevelEnum.WARN.value:
Console.set_foreground_color(ForegroundColorEnum.yellow)
Console.write_line(output)
Console.set_foreground_color(ForegroundColorEnum.default)
def error(self, name: str, message: str, ex: Exception = None):
output = ""
if ex is not None:
tb = traceback.format_exc()
self.error(name, message)
output = self._get_string(name, LoggingLevelEnum.ERROR, f"{ex} -> {tb}")
else:
output = self._get_string(name, LoggingLevelEnum.ERROR, message)
# check if message can be written to log
if self._level.value >= LoggingLevelEnum.ERROR.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevelEnum.ERROR.value:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(output)
Console.set_foreground_color(ForegroundColorEnum.default)
def fatal(self, name: str, message: str, ex: Exception = None):
output = ""
if ex is not None:
tb = traceback.format_exc()
self.error(name, message)
output = self._get_string(name, LoggingLevelEnum.FATAL, f"{ex} -> {tb}")
else:
output = self._get_string(name, LoggingLevelEnum.FATAL, message)
# check if message can be written to log
if self._level.value >= LoggingLevelEnum.FATAL.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevelEnum.FATAL.value:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(output)
Console.set_foreground_color(ForegroundColorEnum.default)
sys.exit()
def _warn_console(self, name: str, message: str):
r"""Writes a warning to console only
Parameter:
name: :class:`str`
Error name
message: :class:`str`
Error message
"""
# check if message can be shown in console
if self._console.value >= LoggingLevelEnum.WARN.value:
Console.set_foreground_color(ForegroundColorEnum.yellow)
Console.write_line(self._get_string(name, LoggingLevelEnum.WARN, message))
Console.set_foreground_color(ForegroundColorEnum.default)
def _fatal_console(self, name: str, message: str, ex: Exception = None):
r"""Writes an error to console only
Parameter:
name: :class:`str`
Error name
message: :class:`str`
Error message
ex: :class:`Exception`
Thrown exception
"""
output = ""
if ex is not None:
tb = traceback.format_exc()
self.error(name, message)
output = self._get_string(name, LoggingLevelEnum.ERROR, f"{ex} -> {tb}")
else:
output = self._get_string(name, LoggingLevelEnum.ERROR, message)
# check if message can be shown in console
if self._console.value >= LoggingLevelEnum.FATAL.value:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(output)
Console.set_foreground_color(ForegroundColorEnum.default)
sys.exit()

11
src/cpl-core/cpl/core/log/logging_level_enum.py
View File

@@ -1,11 +0,0 @@
from enum import Enum
class LoggingLevelEnum(Enum):
OFF = 0 # Nothing
FATAL = 1 # Error that cause exit
ERROR = 2 # Non fatal error
WARN = 3 # Error that can later be fatal
INFO = 4 # Normal information's
DEBUG = 5 # Detailed app state
TRACE = 6 # Detailed app information's

20
src/cpl-core/cpl/core/log/logging_settings.py
View File

@@ -1,24 +1,24 @@
from typing import Optional
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.core.log.logging_level_enum import LoggingLevelEnum
from cpl.core.log.log_level_enum import LogLevelEnum
class LoggingSettings(ConfigurationModelABC):
class LogSettings(ConfigurationModelABC):
r"""Representation of logging settings"""
def __init__(
self,
path: str = None,
filename: str = None,
console_log_level: LoggingLevelEnum = None,
file_log_level: LoggingLevelEnum = None,
console_log_level: LogLevelEnum = None,
file_log_level: LogLevelEnum = None,
):
ConfigurationModelABC.__init__(self)
self._path: Optional[str] = path
self._filename: Optional[str] = filename
self._console: Optional[LoggingLevelEnum] = console_log_level
self._level: Optional[LoggingLevelEnum] = file_log_level
self._console: Optional[LogLevelEnum] = console_log_level
self._level: Optional[LogLevelEnum] = file_log_level
@property
def path(self) -> str:
@@ -37,17 +37,17 @@ class LoggingSettings(ConfigurationModelABC):
self._filename = filename
@property
def console(self) -> LoggingLevelEnum:
def console(self) -> LogLevelEnum:
return self._console
@console.setter
def console(self, console: LoggingLevelEnum) -> None:
def console(self, console: LogLevelEnum) -> None:
self._console = console
@property
def level(self) -> LoggingLevelEnum:
def level(self) -> LogLevelEnum:
return self._level
@level.setter
def level(self, level: LoggingLevelEnum) -> None:
def level(self, level: LogLevelEnum) -> None:
self._level = level

8
src/cpl-core/cpl/core/log/logging_settings_name_enum.py
View File

@@ -1,8 +0,0 @@
from enum import Enum
class LoggingSettingsNameEnum(Enum):
path = "Path"
filename = "Filename"
console_level = "ConsoleLogLevel"
file_level = "FileLogLevel"

5
src/cpl-core/cpl/core/mailing/__init__.py
View File

@@ -1,5 +0,0 @@
from .email import EMail
from .email_client_service import EMailClient
from .email_client_abc import EMailClientABC
from .email_client_settings import EMailClientSettings
from .email_client_settings_name_enum import EMailClientSettingsNameEnum

4
src/cpl-core/cpl/core/type.py
View File

@@ -1,4 +0,0 @@
from typing import TypeVar
T = TypeVar("T")
R = TypeVar("R")

9
src/cpl-core/cpl/core/typing.py Normal file
View File

@@ -0,0 +1,9 @@
from typing import TypeVar, Any
T = TypeVar("T")
R = TypeVar("R")
Service = TypeVar("Service")
Source = TypeVar("Source")
Messages = list[Any] | Any

63
src/cpl-core/cpl/core/utils/get_value.py Normal file
View File

@@ -0,0 +1,63 @@
from typing import Type, Optional
from cpl.core.typing import T
def get_value(
source: dict,
key: str,
cast_type: Type[T],
default: Optional[T] = None,
list_delimiter: str = ",",
) -> Optional[T]:
"""
Get value from source dictionary and cast it to a specified type.
:param dict source: The source dictionary.
:param str key: The name of the environment variable.
:param Type[T] cast_type: A callable to cast the variable's value.
:param Optional[T] default: The default value to return if the variable is not found. Defaults to None.
:param str list_delimiter: The delimiter to split the value into a list. Defaults to ",".
:return: The casted value, or None if the key is not found.
:rtype: Optional[T]
"""
if key not in source:
return default
value = source[key]
if isinstance(
value,
cast_type if not hasattr(cast_type, "__origin__") else cast_type.__origin__,
):
# Handle list[int] case explicitly
if hasattr(cast_type, "__origin__") and cast_type.__origin__ == list:
subtype = cast_type.__args__[0] if hasattr(cast_type, "__args__") else None
if subtype is not None:
return [subtype(item) for item in value]
return value
try:
if cast_type == bool:
return value.lower() in ["true", "1"]
if (
cast_type if not hasattr(cast_type, "__origin__") else cast_type.__origin__
) == list:
if (
not (value.startswith("[") and value.endswith("]"))
and list_delimiter not in value
):
raise ValueError(
"List values must be enclosed in square brackets or use a delimiter."
)
if value.startswith("[") and value.endswith("]"):
value = value[1:-1]
value = value.split(list_delimiter)
subtype = cast_type.__args__[0] if hasattr(cast_type, "__args__") else None
return [subtype(item) if subtype is not None else item for item in value]
return cast_type(value)
except (ValueError, TypeError):
return default

27
src/cpl-mail/cpl/mail/__init__.py
View File

@@ -1,5 +1,26 @@
from .email_model import EMail
from .email_client_service import EMailClient
from .abc.email_client_abc import EMailClientABC
from .email_client import EMailClient
from .email_client_settings import EMailClientSettings
from .email_client_settings_name_enum import EMailClientSettingsNameEnum
from .email_client_settings_name_enum import EMailClientSettingsNameEnum
from .email_model import EMail
from .mail_logger import MailLogger
def add_mail(self):
from cpl.core.console import Console
from cpl.core.log import LoggerABC
try:
self.add_singleton(EMailClientABC, EMailClient)
self.add_transient(LoggerABC, MailLogger)
except ImportError as e:
Console.error("cpl-translation is not installed", str(e))
def init():
from cpl.core.dependency_injection import ServiceCollection
ServiceCollection.add_mail = add_mail
init()

40
src/cpl-mail/cpl/mail/email_client_service.py → src/cpl-mail/cpl/mail/email_client.py
View File

@@ -3,11 +3,11 @@ from smtplib import SMTP
from typing import Optional
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.utils.credential_manager import CredentialManager
from cpl.mail.abc.email_client_abc import EMailClientABC
from cpl.mail.email_model import EMail
from cpl.mail.email_client_settings import EMailClientSettings
from cpl.mail.email_model import EMail
from cpl.mail.mail_logger import MailLogger
class EMailClient(EMailClientABC):
@@ -22,7 +22,7 @@ class EMailClient(EMailClientABC):
Settings for mailing
"""
def __init__(self, environment: ApplicationEnvironmentABC, logger: LoggerABC, mail_settings: EMailClientSettings):
def __init__(self, environment: ApplicationEnvironmentABC, logger: MailLogger, mail_settings: EMailClientSettings):
EMailClientABC.__init__(self)
self._environment = environment
@@ -35,28 +35,28 @@ class EMailClient(EMailClientABC):
def create(self):
r"""Creates connection"""
self._logger.trace(__name__, f"Started {__name__}.create")
self._logger.trace(f"Started {__name__}.create")
self.connect()
self._logger.trace(__name__, f"Stopped {__name__}.create")
self._logger.trace(f"Stopped {__name__}.create")
def connect(self):
self._logger.trace(__name__, f"Started {__name__}.connect")
self._logger.trace(f"Started {__name__}.connect")
try:
self._logger.debug(__name__, f"Try to connect to {self._mail_settings.host}:{self._mail_settings.port}")
self._logger.debug(f"Try to connect to {self._mail_settings.host}:{self._mail_settings.port}")
self._server = SMTP(self._mail_settings.host, self._mail_settings.port)
self._logger.info(__name__, f"Connected to {self._mail_settings.host}:{self._mail_settings.port}")
self._logger.info(f"Connected to {self._mail_settings.host}:{self._mail_settings.port}")
self._logger.debug(__name__, "Try to start tls")
self._logger.debug("Try to start tls")
self._server.starttls(context=ssl.create_default_context())
self._logger.info(__name__, "Started tls")
self._logger.info("Started tls")
except Exception as e:
self._logger.error(__name__, "Cannot connect to mail server", e)
self._logger.error("Cannot connect to mail server", e)
self._logger.trace(__name__, f"Stopped {__name__}.connect")
self._logger.trace(f"Stopped {__name__}.connect")
def login(self):
r"""Login to server"""
self._logger.trace(__name__, f"Started {__name__}.login")
self._logger.trace(f"Started {__name__}.login")
try:
self._logger.debug(
__name__,
@@ -70,19 +70,19 @@ class EMailClient(EMailClientABC):
f"Logged on as {self._mail_settings.user_name} to {self._mail_settings.host}:{self._mail_settings.port}",
)
except Exception as e:
self._logger.error(__name__, "Cannot login to mail server", e)
self._logger.error("Cannot login to mail server", e)
self._logger.trace(__name__, f"Stopped {__name__}.login")
self._logger.trace(f"Stopped {__name__}.login")
def send_mail(self, email: EMail):
self._logger.trace(__name__, f"Started {__name__}.send_mail")
self._logger.trace(f"Started {__name__}.send_mail")
try:
self.login()
self._logger.debug(__name__, f"Try to send email to {email.receiver_list}")
self._logger.debug(f"Try to send email to {email.receiver_list}")
self._server.sendmail(
self._mail_settings.user_name, email.receiver_list, email.get_content(self._mail_settings.user_name)
)
self._logger.info(__name__, f"Sent email to {email.receiver_list}")
self._logger.info(f"Sent email to {email.receiver_list}")
except Exception as e:
self._logger.error(__name__, f"Cannot send mail to {email.receiver_list}", e)
self._logger.trace(__name__, f"Stopped {__name__}.send_mail")
self._logger.error(f"Cannot send mail to {email.receiver_list}", e)
self._logger.trace(f"Stopped {__name__}.send_mail")

8
src/cpl-mail/cpl/mail/mail_logger.py Normal file
View File

@@ -0,0 +1,8 @@
from cpl.core.log.logger import Logger
from cpl.core.typing import Source
class MailLogger(Logger):
def __init__(self, source: Source):
Logger.__init__(self, source, "mail")