Compare commits

...

5 Commits

Author SHA1 Message Date
58dbd3ed1e Cleanup for mysql
All checks were successful
Build on push / prepare (push) Successful in 8s
Build on push / core (push) Successful in 19s
Build on push / query (push) Successful in 18s
Build on push / translation (push) Successful in 15s
Build on push / mail (push) Successful in 17s
2025-09-16 20:21:33 +02:00
cd7dfaf2b4 Fixed spinner thread & remove log thread
All checks were successful
Build on push / prepare (push) Successful in 8s
Build on push / core (push) Successful in 18s
Build on push / query (push) Successful in 19s
Build on push / translation (push) Successful in 18s
Build on push / mail (push) Successful in 18s
2025-09-16 18:41:57 +02:00
8ad3e3bdb4 Removed ConfigModel from_dict
All checks were successful
Build on push / prepare (push) Successful in 8s
Build on push / core (push) Successful in 16s
Build on push / query (push) Successful in 17s
Build on push / mail (push) Successful in 17s
Build on push / translation (push) Successful in 18s
2025-09-16 08:51:56 +02:00
b97bc0a3ed Restructuring
All checks were successful
Build on push / prepare (push) Successful in 8s
Build on push / query (push) Successful in 17s
Build on push / core (push) Successful in 26s
Build on push / translation (push) Successful in 15s
Build on push / mail (push) Successful in 17s
2025-09-16 08:48:07 +02:00
5f25400bcd Updated config & environment 2025-09-16 08:39:00 +02:00
198 changed files with 847 additions and 2807 deletions

View File

@@ -1,10 +1,9 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.dependency.service_provider_abc import ServiceProviderABC
from cpl.core.console.console import Console
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
class ApplicationABC(ABC):
@@ -13,14 +12,12 @@ class ApplicationABC(ABC):
Parameters:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
Contains object loaded from appsettings
services: :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
services: :class:`cpl.dependency.service_provider_abc.ServiceProviderABC`
Contains instances of prepared objects
"""
@abstractmethod
def __init__(self, config: ConfigurationABC, services: ServiceProviderABC):
self._configuration: Optional[ConfigurationABC] = config
self._environment: Optional[ApplicationEnvironmentABC] = self._configuration.environment
def __init__(self, services: ServiceProviderABC):
self._services: Optional[ServiceProviderABC] = services
def run(self):
@@ -49,12 +46,12 @@ class ApplicationABC(ABC):
def configure(self):
r"""Configure the application
Called by :class:`cpl.core.application.application_abc.ApplicationABC.run`
Called by :class:`cpl.application.application_abc.ApplicationABC.run`
"""
@abstractmethod
def main(self):
r"""Custom entry point
Called by :class:`cpl.core.application.application_abc.ApplicationABC.run`
Called by :class:`cpl.application.application_abc.ApplicationABC.run`
"""

View File

@@ -0,0 +1,97 @@
from typing import Type, Optional, Callable, Union
from cpl.application.application_abc import ApplicationABC
from cpl.application.application_builder_abc import ApplicationBuilderABC
from cpl.application.application_extension_abc import ApplicationExtensionABC
from cpl.application.async_application_extension_abc import AsyncApplicationExtensionABC
from cpl.application.async_startup_abc import AsyncStartupABC
from cpl.application.async_startup_extension_abc import AsyncStartupExtensionABC
from cpl.application.startup_abc import StartupABC
from cpl.application.startup_extension_abc import StartupExtensionABC
from cpl.core.configuration.configuration import Configuration
from cpl.dependency.service_collection import ServiceCollection
from cpl.core.environment import Environment
class ApplicationBuilder(ApplicationBuilderABC):
r"""This is class is used to build an object of :class:`cpl.application.application_abc.ApplicationABC`
Parameter:
app: Type[:class:`cpl.application.application_abc.ApplicationABC`]
Application to build
"""
def __init__(self, app: Type[ApplicationABC]):
ApplicationBuilderABC.__init__(self)
self._app = app
self._startup: Optional[StartupABC | AsyncStartupABC] = None
self._services = ServiceCollection()
self._app_extensions: list[Type[ApplicationExtensionABC | AsyncApplicationExtensionABC]] = []
self._startup_extensions: list[Type[StartupExtensionABC | AsyncStartupABC]] = []
def use_startup(self, startup: Type[StartupABC | AsyncStartupABC]) -> "ApplicationBuilder":
self._startup = startup()
return self
def use_extension(
self,
extension: Type[
ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC
],
) -> "ApplicationBuilder":
if (
issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)
) and extension not in self._app_extensions:
self._app_extensions.append(extension)
elif (
issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)
) and extension not in self._startup_extensions:
self._startup_extensions.append(extension)
return self
def _build_startup(self):
for ex in self._startup_extensions:
extension = ex()
extension.configure_configuration(Configuration, Environment)
extension.configure_services(self._services, Environment)
if self._startup is not None:
self._startup.configure_configuration(Configuration, Environment)
self._startup.configure_services(self._services, Environment)
async def _build_async_startup(self):
for ex in self._startup_extensions:
extension = ex()
await extension.configure_configuration(Configuration, Environment)
await extension.configure_services(self._services, Environment)
if self._startup is not None:
await self._startup.configure_configuration(Configuration, Environment)
await self._startup.configure_services(self._services, Environment)
def build(self) -> ApplicationABC:
self._build_startup()
config = Configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
extension.run(config, services)
return self._app(services)
async def build_async(self) -> ApplicationABC:
await self._build_async_startup()
config = Configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
await extension.run(config, services)
return self._app(services)

View File

@@ -1,12 +1,12 @@
from abc import ABC, abstractmethod
from typing import Type
from cpl.core.application.application_abc import ApplicationABC
from cpl.core.application.startup_abc import StartupABC
from cpl.application.application_abc import ApplicationABC
from cpl.application.startup_abc import StartupABC
class ApplicationBuilderABC(ABC):
r"""ABC for the :class:`cpl.core.application.application_builder.ApplicationBuilder`"""
r"""ABC for the :class:`cpl.application.application_builder.ApplicationBuilder`"""
@abstractmethod
def __init__(self, *args):
@@ -17,7 +17,7 @@ class ApplicationBuilderABC(ABC):
r"""Sets the custom startup class to use
Parameter:
startup: Type[:class:`cpl.core.application.startup_abc.StartupABC`]
startup: Type[:class:`cpl.application.startup_abc.StartupABC`]
Startup class to use
"""
@@ -26,7 +26,7 @@ class ApplicationBuilderABC(ABC):
r"""Sets the custom startup class to use async
Parameter:
startup: Type[:class:`cpl.core.application.startup_abc.StartupABC`]
startup: Type[:class:`cpl.application.startup_abc.StartupABC`]
Startup class to use
"""
@@ -35,7 +35,7 @@ class ApplicationBuilderABC(ABC):
r"""Creates custom application object
Returns:
Object of :class:`cpl.core.application.application_abc.ApplicationABC`
Object of :class:`cpl.application.application_abc.ApplicationABC`
"""
@abstractmethod
@@ -43,5 +43,5 @@ class ApplicationBuilderABC(ABC):
r"""Creates custom application object async
Returns:
Object of :class:`cpl.core.application.application_abc.ApplicationABC`
Object of :class:`cpl.application.application_abc.ApplicationABC`
"""

View File

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

View File

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

View File

@@ -0,0 +1,23 @@
from abc import ABC, abstractmethod
from cpl.dependency.service_collection import ServiceCollection
class AsyncStartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def configure_configuration(self):
r"""Creates configuration of application"""
@abstractmethod
async def configure_services(self, service: ServiceCollection):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection`
"""

View File

@@ -1,8 +1,8 @@
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
from cpl.core.configuration.configuration import Configuration
from cpl.dependency.service_collection import ServiceCollection
from cpl.core.environment.environment import Environment
class AsyncStartupExtensionABC(ABC):
@@ -13,19 +13,19 @@ class AsyncStartupExtensionABC(ABC):
pass
@abstractmethod
async def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
async def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
config: :class:`cpl.core.configuration.configuration_abc.Configuration`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
async def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
async def configure_services(self, service: ServiceCollection, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
services: :class:`cpl.dependency.service_collection`
env: :class:`cpl.core.environment.application_environment_abc`
"""

View File

@@ -0,0 +1,31 @@
from abc import ABC, abstractmethod
from cpl.core.configuration import Configuration
from cpl.dependency.service_collection import ServiceCollection
from cpl.core.environment import Environment
class StartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
def configure_services(self, service: ServiceCollection, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection`
env: :class:`cpl.core.environment.application_environment_abc`
"""

View File

@@ -1,8 +1,10 @@
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
from cpl.core.configuration import Configuration
from cpl.dependency.service_collection import ServiceCollection
from cpl.core.environment.environment import Environment
class StartupExtensionABC(ABC):
@@ -13,7 +15,7 @@ class StartupExtensionABC(ABC):
pass
@abstractmethod
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
@@ -22,10 +24,10 @@ class StartupExtensionABC(ABC):
"""
@abstractmethod
def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
def configure_services(self, service: ServiceCollection, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
services: :class:`cpl.dependency.service_collection`
env: :class:`cpl.core.environment.application_environment_abc`
"""

View File

@@ -0,0 +1,30 @@
[build-system]
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
build-backend = "setuptools.build_meta"
[project]
name = "cpl-application"
version = "2024.7.0"
description = "CPL application"
readme ="CPL application package"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
]
keywords = ["cpl", "application", "backend", "shared", "library"]
dynamic = ["dependencies", "optional-dependencies"]
[project.urls]
Homepage = "https://www.sh-edraft.de"
[tool.setuptools.packages.find]
where = ["."]
include = ["cpl*"]
[tool.setuptools.dynamic]
dependencies = { file = ["requirements.txt"] }
optional-dependencies.dev = { file = ["requirements.dev.txt"] }

View File

@@ -0,0 +1 @@
black==25.1.0

View File

@@ -0,0 +1,2 @@
cpl-core
cpl-dependency

View File

@@ -1,91 +0,0 @@
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
from cpl.core.dependency_injection.service_collection import ServiceCollection
class ApplicationBuilder(ApplicationBuilderABC):
r"""This is class is used to build an object of :class:`cpl.core.application.application_abc.ApplicationABC`
Parameter:
app: Type[:class:`cpl.core.application.application_abc.ApplicationABC`]
Application to build
"""
def __init__(self, app: Type[ApplicationABC]):
ApplicationBuilderABC.__init__(self)
self._app = app
self._startup: Optional[StartupABC | AsyncStartupABC] = None
self._configuration = Configuration()
self._environment = self._configuration.environment
self._services = ServiceCollection(self._configuration)
self._app_extensions: list[Type[ApplicationExtensionABC | AsyncApplicationExtensionABC]] = []
self._startup_extensions: list[Type[StartupExtensionABC | AsyncStartupABC]] = []
def use_startup(self, startup: Type[StartupABC | AsyncStartupABC]) -> "ApplicationBuilder":
self._startup = startup()
return self
def use_extension(
self, extension: Type[ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC]
) -> "ApplicationBuilder":
if (issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)) and extension not in self._app_extensions:
self._app_extensions.append(extension)
elif (issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)) and extension not in self._startup_extensions:
self._startup_extensions.append(extension)
return self
def _build_startup(self):
for ex in self._startup_extensions:
extension = ex()
extension.configure_configuration(self._configuration, self._environment)
extension.configure_services(self._services, self._environment)
if self._startup is not None:
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()
config = self._configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
extension.run(config, services)
return self._app(config, services)
async def build_async(self) -> ApplicationABC:
await self._build_async_startup()
config = self._configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
await extension.run(config, services)
return self._app(config, services)

View File

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

View File

@@ -1,14 +0,0 @@
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

View File

@@ -1,32 +0,0 @@
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`
"""

View File

@@ -1,32 +0,0 @@
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 StartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
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
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`
"""

View File

@@ -1,12 +1,2 @@
from .argument_abc import ArgumentABC
from .argument_builder import ArgumentBuilder
from .argument_executable_abc import ArgumentExecutableABC
from .argument_type_enum import ArgumentTypeEnum
from .configuration import Configuration
from .configuration_abc import ConfigurationABC
from .configuration_model_abc import ConfigurationModelABC
from .configuration_variable_name_enum import ConfigurationVariableNameEnum
from .executable_argument import ExecutableArgument
from .flag_argument import FlagArgument
from .validator_abc import ValidatorABC
from .variable_argument import VariableArgument

View File

@@ -1,64 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.argument_type_enum import ArgumentTypeEnum
class ArgumentABC(ABC):
@abstractmethod
def __init__(
self,
token: str,
name: str,
aliases: list[str],
prevent_next_executable: bool = False,
console_arguments: list["ArgumentABC"] = None,
):
r"""Representation of an console argument
Parameter:
token: :class:`str`
name: :class:`str`
aliases: list[:class:`str`]
console_arguments: List[:class:`cpl.core.configuration.console_argument.ConsoleArgument`]
"""
self._token = token
self._name = name
self._aliases = aliases
self._prevent_next_executable = prevent_next_executable
self._console_arguments = console_arguments if console_arguments is not None else []
@property
def token(self) -> str:
return self._token
@property
def name(self) -> str:
return self._name
@property
def aliases(self) -> list[str]:
return self._aliases
@property
def prevent_next_executable(self) -> bool:
return self._prevent_next_executable
@property
def console_arguments(self) -> list["ArgumentABC"]:
return self._console_arguments
def add_console_argument(self, arg_type: ArgumentTypeEnum, *args, **kwargs) -> "ArgumentABC":
r"""Creates and adds a console argument to known console arguments
Parameter:
arg_type: :class:`str`
Specifies the specific type of the argument
Returns:
self :class:`cpl.core.configuration.console_argument.ConsoleArgument` not created argument!
"""
from cpl.core.configuration.argument_builder import ArgumentBuilder
argument = ArgumentBuilder.build_argument(arg_type, *args, *kwargs)
self._console_arguments.append(argument)
return self

View File

@@ -1,30 +0,0 @@
from typing import Union
from cpl.core.configuration.argument_type_enum import ArgumentTypeEnum
from cpl.core.configuration.executable_argument import ExecutableArgument
from cpl.core.configuration.flag_argument import FlagArgument
from cpl.core.configuration.variable_argument import VariableArgument
from cpl.core.console import Console
class ArgumentBuilder:
@staticmethod
def build_argument(
arg_type: ArgumentTypeEnum, *args, **kwargs
) -> Union[ExecutableArgument, FlagArgument, VariableArgument]:
argument = None
try:
match arg_type:
case ArgumentTypeEnum.Flag:
argument = FlagArgument(*args, **kwargs)
case ArgumentTypeEnum.Executable:
argument = ExecutableArgument(*args, **kwargs)
case ArgumentTypeEnum.Variable:
argument = VariableArgument(*args, **kwargs)
case _:
Console.error("Invalid argument type")
Console.close()
except TypeError as e:
Console.error(str(e))
Console.close()
return argument

View File

@@ -1,11 +0,0 @@
from abc import ABC, abstractmethod
class ArgumentExecutableABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def run(self, args: list[str]):
pass

View File

@@ -1,7 +0,0 @@
from enum import Enum
class ArgumentTypeEnum(Enum):
Flag = 0
Executable = 1
Variable = 3

View File

@@ -1,65 +1,19 @@
import inspect
import json
import os
import sys
import traceback
from collections.abc import Callable
from typing import Union, Optional
from typing import Any
from cpl.core.configuration.argument_abc import ArgumentABC
from cpl.core.configuration.argument_builder import ArgumentBuilder
from cpl.core.configuration.argument_type_enum import ArgumentTypeEnum
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.core.configuration.configuration_variable_name_enum import (
ConfigurationVariableNameEnum,
)
from cpl.core.configuration.executable_argument import ExecutableArgument
from cpl.core.configuration.flag_argument import FlagArgument
from cpl.core.configuration.variable_argument import VariableArgument
from cpl.core.console.console import Console
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
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.typing import T, R
from cpl.core.environment.environment import Environment
from cpl.core.typing import D, T
from cpl.core.utils.json_processor import JSONProcessor
class Configuration(ConfigurationABC):
def __init__(self):
r"""Representation of configuration"""
ConfigurationABC.__init__(self)
self._application_environment = ApplicationEnvironment()
self._config: dict[Union[type, str], Union[ConfigurationModelABC, str]] = {}
self._argument_types: list[ArgumentABC] = []
self._additional_arguments: list[str] = []
self._argument_error_function: Optional[Callable] = None
self._handled_args = []
@property
def environment(self) -> ApplicationEnvironmentABC:
return self._application_environment
@property
def additional_arguments(self) -> list[str]:
return self._additional_arguments
@property
def argument_error_function(self) -> Optional[Callable]:
return self._argument_error_function
@argument_error_function.setter
def argument_error_function(self, argument_error_function: Callable):
self._argument_error_function = argument_error_function
@property
def arguments(self) -> list[ArgumentABC]:
return self._argument_types
class Configuration:
_config = {}
@staticmethod
def _print_info(message: str):
@@ -72,7 +26,7 @@ class Configuration(ConfigurationABC):
Info message
"""
Console.set_foreground_color(ForegroundColorEnum.green)
Console.write_line(f"[CONF] {message}")
Console.write_line(f"[CONFIG] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
@staticmethod
@@ -86,7 +40,7 @@ class Configuration(ConfigurationABC):
Warning message
"""
Console.set_foreground_color(ForegroundColorEnum.yellow)
Console.write_line(f"[CONF] {message}")
Console.write_line(f"[CONFIG] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
@staticmethod
@@ -100,31 +54,11 @@ class Configuration(ConfigurationABC):
Error message
"""
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f"[CONF] {message}")
Console.write_line(f"[CONFIG] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
def _set_variable(self, name: str, value: any):
r"""Sets variable to given value
Parameter:
name: :class:`str`
Name of the variable
value: :class:`any`
Value of the variable
"""
if name == ConfigurationVariableNameEnum.environment.value:
self._application_environment.environment_name = EnvironmentNameEnum(value)
elif name == ConfigurationVariableNameEnum.name.value:
self._application_environment.application_name = value
elif name == ConfigurationVariableNameEnum.customer.value:
self._application_environment.customer = value
else:
self._config[name] = value
def _load_json_file(self, file: str, output: bool) -> dict:
@classmethod
def _load_json_file(cls, file: str, output: bool) -> dict:
r"""Reads the json file
Parameter:
@@ -142,90 +76,19 @@ class Configuration(ConfigurationABC):
# load json
json_cfg = json.load(cfg)
if output:
self._print_info(f"Loaded config file: {file}")
cls._print_info(f"Loaded config file: {file}")
return json_cfg
except Exception as e:
self._print_error(f"Cannot load config file: {file}! -> {e}")
cls._print_error(f"Cannot load config file: {file}! -> {e}")
return {}
def _parse_arguments(
self,
executables: list[ArgumentABC],
arg_list: list[str],
args_types: list[ArgumentABC],
):
for i in range(0, len(arg_list)):
arg_str = arg_list[i]
for n in range(0, len(args_types)):
arg = args_types[n]
arg_str_without_token = arg_str
if arg.token != "" and arg.token in arg_str:
arg_str_without_token = arg_str.split(arg.token)[1]
# executable
if isinstance(arg, ExecutableArgument):
if (
arg_str.startswith(arg.token)
and arg_str_without_token == arg.name
or arg_str_without_token in arg.aliases
):
executables.append(arg)
self._handled_args.append(arg_str)
self._parse_arguments(executables, arg_list[i + 1 :], arg.console_arguments)
# variables
elif isinstance(arg, VariableArgument):
arg_str_without_value = arg_str_without_token
if arg.value_token in arg_str_without_value:
arg_str_without_value = arg_str_without_token.split(arg.value_token)[0]
if (
arg_str.startswith(arg.token)
and arg_str_without_value == arg.name
or arg_str_without_value in arg.aliases
):
if arg.value_token != " ":
value = arg_str_without_token.split(arg.value_token)[1]
else:
value = arg_list[i + 1]
self._set_variable(arg.name, value)
self._handled_args.append(arg_str)
self._handled_args.append(value)
self._parse_arguments(executables, arg_list[i + 1 :], arg.console_arguments)
# flags
elif isinstance(arg, FlagArgument):
if (
arg_str.startswith(arg.token)
and arg_str_without_token == arg.name
or arg_str_without_token in arg.aliases
):
if arg_str in self._additional_arguments:
self._additional_arguments.remove(arg_str)
self._additional_arguments.append(arg.name)
self._handled_args.append(arg_str)
self._parse_arguments(executables, arg_list[i + 1 :], arg.console_arguments)
# add left over values to args
if arg_str not in self._additional_arguments and arg_str not in self._handled_args:
self._additional_arguments.append(arg_str)
def add_environment_variables(self, prefix: str):
for env_var in os.environ.keys():
if not env_var.startswith(prefix):
continue
self._set_variable(env_var.replace(prefix, ""), os.environ[env_var])
def add_console_argument(self, argument: ArgumentABC):
self._argument_types.append(argument)
def add_json_file(self, name: str, optional: bool = None, output: bool = True, path: str = None):
@classmethod
def add_json_file(cls, name: str, optional: bool = None, output: bool = True, path: str = None):
if os.path.isabs(name):
file_path = name
else:
path_root = self._application_environment.working_directory
path_root = Environment.get_cwd()
if path is not None:
path_root = path
@@ -237,122 +100,35 @@ class Configuration(ConfigurationABC):
if not os.path.isfile(file_path):
if optional is not True:
if output:
self._print_error(f"File not found: {file_path}")
cls._print_error(f"File not found: {file_path}")
sys.exit()
if output:
self._print_warn(__name__, f"Not Loaded config file: {file_path}")
cls._print_warn(f"Not Loaded config file: {file_path}")
return None
config_from_file = self._load_json_file(file_path, output)
config_from_file = cls._load_json_file(file_path, output)
for sub in ConfigurationModelABC.__subclasses__():
for key, value in config_from_file.items():
if sub.__name__ != key and sub.__name__.replace("Settings", "") != key:
continue
configuration = sub()
from_dict = getattr(configuration, "from_dict", None)
configuration = JSONProcessor.process(sub, value)
if from_dict is not None and not hasattr(from_dict, "is_base_func"):
Console.set_foreground_color(ForegroundColorEnum.yellow)
Console.write_line(
f"{sub.__name__}.from_dict is deprecated. Instead, set attributes as typed arguments in __init__. They can be None by default!"
)
Console.color_reset()
configuration.from_dict(value)
else:
configuration = JSONProcessor.process(sub, value)
cls.set(sub, configuration)
self.add_configuration(sub, configuration)
@classmethod
def set(cls, key: Any, value: T):
if inspect.isclass(key):
key = key.__name__
def add_configuration(self, key_type: T, value: any):
self._config[key_type] = value
cls._config[key] = value
def create_console_argument(
self,
arg_type: ArgumentTypeEnum,
token: str,
name: str,
aliases: list[str],
*args,
**kwargs,
) -> ArgumentABC:
argument = ArgumentBuilder.build_argument(arg_type, token, name, aliases, *args, **kwargs)
self._argument_types.append(argument)
return argument
@classmethod
def get(cls, key: Any, default: D = None) -> T | D:
if inspect.isclass(key):
key = key.__name__
def for_each_argument(self, call: Callable):
for arg in self._argument_types:
call(arg)
def get_configuration(self, search_type: T) -> Optional[R]:
if type(search_type) is str:
if search_type == ConfigurationVariableNameEnum.environment.value:
return self._application_environment.environment_name
elif search_type == ConfigurationVariableNameEnum.name.value:
return self._application_environment.application_name
elif search_type == ConfigurationVariableNameEnum.customer.value:
return self._application_environment.customer
if search_type not in self._config:
return None
for config_model in self._config:
if config_model == search_type:
return self._config[config_model]
def parse_console_arguments(self, services: ServiceProviderABC, error: bool = None) -> bool:
# sets environment variables as possible arguments as: --VAR=VALUE
for arg_name in ConfigurationVariableNameEnum.to_list():
self.add_console_argument(VariableArgument("--", str(arg_name).upper(), [str(arg_name).lower()], "="))
success = False
try:
arg_list = sys.argv[1:]
executables: list[ExecutableArgument] = []
self._parse_arguments(executables, arg_list, self._argument_types)
except Exception:
Console.error("An error occurred while parsing arguments.", traceback.format_exc())
sys.exit()
try:
prevent = False
for exe in executables:
if prevent:
continue
if exe.validators is not None:
abort = False
for validator_type in exe.validators:
validator = services.get_service(validator_type)
result = validator.validate()
abort = not result
if abort:
break
if abort:
sys.exit()
cmd = services.get_service(exe.executable_type)
self._handle_pre_or_post_executables(True, exe, services)
self._set_variable("ACTIVE_EXECUTABLE", exe.name)
args = self.get_configuration("ARGS")
if args is not None:
for arg in args.split(" "):
if arg == "":
continue
self._additional_arguments.append(arg)
cmd.run(self._additional_arguments)
self._handle_pre_or_post_executables(False, exe, services)
prevent = exe.prevent_next_executable
success = True
except Exception:
Console.error("An error occurred while executing arguments.", traceback.format_exc())
sys.exit()
return success
return cls._config.get(key, default)

View File

@@ -1,140 +0,0 @@
from abc import abstractmethod, ABC
from collections.abc import Callable
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.typing import T, R
class ConfigurationABC(ABC):
@abstractmethod
def __init__(self):
r"""ABC for the :class:`cpl.core.configuration.configuration.Configuration`"""
@property
@abstractmethod
def environment(self) -> ApplicationEnvironmentABC:
pass
@property
@abstractmethod
def additional_arguments(self) -> list[str]:
pass
@property
@abstractmethod
def argument_error_function(self) -> Optional[Callable]:
pass
@argument_error_function.setter
@abstractmethod
def argument_error_function(self, argument_error_function: Callable):
pass
@property
@abstractmethod
def arguments(self) -> list[ArgumentABC]:
pass
@abstractmethod
def add_environment_variables(self, prefix: str):
r"""Reads the environment variables
Parameter:
prefix: :class:`str`
Prefix of the variables
"""
@abstractmethod
def add_console_argument(self, argument: ArgumentABC):
r"""Adds console argument to known console arguments
Parameter:
argument: :class:`cpl.core.configuration.console_argument.ConsoleArgumentABC`
Specifies the console argument
"""
@abstractmethod
def add_json_file(self, name: str, optional: bool = None, output: bool = True, path: str = None):
r"""Reads and saves settings from given json file
Parameter:
name: :class:`str`
Name of the file
optional: :class:`str`
Specifies whether an error should occur if the file was not found
output: :class:`bool`
Specifies whether an output should take place
path: :class:`str`
Path in which the file should be stored
"""
@abstractmethod
def add_configuration(self, key_type: T, value: any):
r"""Add configuration object
Parameter:
key_type: :class:`cpl.core.type.T`
Type of the value
value: any
Object of the value
"""
@abstractmethod
def create_console_argument(
self, arg_type: ArgumentTypeEnum, token: str, name: str, aliases: list[str], *args, **kwargs
) -> ArgumentABC:
r"""Creates and adds a console argument to known console arguments
Parameter:
token: :class:`str`
Specifies optional beginning of argument
name :class:`str`
Specifies name of argument
aliases list[:class:`str`]
Specifies possible aliases of name
value_token :class:`str`
Specifies were the value begins
is_value_token_optional :class:`bool`
Specifies if values are optional
runnable: :class:`cpl.core.configuration.console_argument.ConsoleArgumentABC`
Specifies class to run when called if value is not None
Returns:
Object of :class:`cpl.core.configuration.console_argument.ConsoleArgumentABC`
"""
@abstractmethod
def for_each_argument(self, call: Callable):
r"""Iterates through all arguments and calls the call function
Parameter:
call: :class:`Callable`
Call for each argument
"""
@abstractmethod
def get_configuration(self, search_type: T) -> Optional[R]:
r"""Returns value from configuration by given type
Parameter:
search_type: :class:`cpl.core.type.T`
Type to search for
Returns:
Object of Union[:class:`str`, :class:`cpl.core.configuration.configuration_model_abc.ConfigurationModelABC`]
"""
@abstractmethod
def parse_console_arguments(self, services: "ServiceProviderABC", error: bool = None) -> bool:
r"""Reads the console arguments
Parameter:
error: :class:`bool`
Defines is invalid argument error will be shown or not
Returns:
Bool to specify if executables were executed or not.
"""

View File

@@ -1,21 +1,5 @@
from abc import ABC, abstractmethod
def base_func(method):
method.is_base_func = True
return method
from abc import ABC
class ConfigurationModelABC(ABC):
@abstractmethod
def __init__(self):
r"""ABC for settings representation"""
@base_func
def from_dict(self, settings: dict):
r"""DEPRECATED: Set attributes as typed arguments in __init__ instead. See https://docs.sh-edraft.de/cpl/deprecated.html#ConfigurationModelABC-from_dict-method for further information
Converts attributes to dict
Parameter:
settings: :class:`dict`
"""
pass

View File

@@ -1,11 +0,0 @@
from enum import Enum
class ConfigurationVariableNameEnum(Enum):
environment = "ENVIRONMENT"
name = "NAME"
customer = "CUSTOMER"
@staticmethod
def to_list():
return [var.value for var in ConfigurationVariableNameEnum]

View File

@@ -1,40 +0,0 @@
from typing import Type, Optional
from cpl.core.configuration.argument_executable_abc import ArgumentExecutableABC
from cpl.core.configuration.argument_abc import ArgumentABC
from cpl.core.configuration.validator_abc import ValidatorABC
class ExecutableArgument(ArgumentABC):
def __init__(
self,
token: str,
name: str,
aliases: list[str],
executable: Type[ArgumentExecutableABC],
prevent_next_executable: bool = False,
validators: list[Type[ValidatorABC]] = None,
console_arguments: list["ArgumentABC"] = None,
):
self._executable_type = executable
self._validators = validators
self._executable: Optional[ArgumentExecutableABC] = None
ArgumentABC.__init__(self, token, name, aliases, prevent_next_executable, console_arguments)
@property
def executable_type(self) -> type:
return self._executable_type
def set_executable(self, executable: ArgumentExecutableABC):
self._executable = executable
@property
def validators(self) -> list[Type[ValidatorABC]]:
return self._validators
def run(self, args: list[str]):
r"""Executes runnable if exists"""
if self._executable is None:
return
self._executable.execute(args)

View File

@@ -1,13 +0,0 @@
from cpl.core.configuration.argument_abc import ArgumentABC
class FlagArgument(ArgumentABC):
def __init__(
self,
token: str,
name: str,
aliases: list[str],
prevent_next_executable: bool = False,
console_arguments: list["ArgumentABC"] = None,
):
ArgumentABC.__init__(self, token, name, aliases, prevent_next_executable, console_arguments)

View File

@@ -1,11 +0,0 @@
from abc import ABC, abstractmethod
class ValidatorABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def validate(self) -> bool:
pass

View File

@@ -1,28 +0,0 @@
from cpl.core.configuration.argument_abc import ArgumentABC
class VariableArgument(ArgumentABC):
def __init__(
self,
token: str,
name: str,
aliases: list[str],
value_token: str,
prevent_next_executable: bool = False,
console_arguments: list["ArgumentABC"] = None,
):
self._value_token = value_token
self._value: str = ""
ArgumentABC.__init__(self, token, name, aliases, prevent_next_executable, console_arguments)
@property
def value_token(self) -> str:
return self._value_token
@property
def value(self) -> str:
return self._value
def set_value(self, value: str):
self._value = value

View File

@@ -1,5 +1,4 @@
from .background_color_enum import BackgroundColorEnum
from .console import Console
from .console_call import ConsoleCall
from ._call import ConsoleCall
from .foreground_color_enum import ForegroundColorEnum
from .spinner_thread import SpinnerThread

View File

@@ -1,7 +1,8 @@
import os
import sys
import threading
import multiprocessing
import time
from multiprocessing import Process
from termcolor import colored
@@ -9,8 +10,8 @@ from cpl.core.console.background_color_enum import BackgroundColorEnum
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
class SpinnerThread(threading.Thread):
r"""Thread to show spinner in terminal
class Spinner(Process):
r"""Process to show spinner in terminal
Parameter:
msg_len: :class:`int`
@@ -22,7 +23,7 @@ class SpinnerThread(threading.Thread):
"""
def __init__(self, msg_len: int, foreground_color: ForegroundColorEnum, background_color: BackgroundColorEnum):
threading.Thread.__init__(self)
Process.__init__(self)
self._msg_len = msg_len
self._foreground_color = foreground_color
@@ -50,29 +51,26 @@ class SpinnerThread(threading.Thread):
return color_args
def run(self) -> None:
r"""Entry point of thread, shows the spinner"""
r"""Entry point of process, shows the spinner"""
columns = 0
if sys.platform == "win32":
columns = os.get_terminal_size().columns
else:
term_rows, term_columns = os.popen("stty size", "r").read().split()
values = os.popen("stty size", "r").read().split()
term_rows, term_columns = values if len(values) == 2 else (0, 0)
columns = int(term_columns)
end_msg = "done"
end_msg_pos = columns - self._msg_len - len(end_msg)
if end_msg_pos > 0:
print(f'{"" : >{end_msg_pos}}', end="")
padding = columns - self._msg_len - len(end_msg)
if padding > 0:
print(f'{"" : >{padding}}', end="")
else:
print("", end="")
first = True
spinner = self._spinner()
while self._is_spinning:
if first:
first = False
print(colored(f"{next(spinner): >{len(end_msg) - 1}}", *self._get_color_args()), end="")
else:
print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
time.sleep(0.1)
back = ""
for i in range(0, len(end_msg)):
@@ -84,9 +82,10 @@ class SpinnerThread(threading.Thread):
if not self._exit:
print(colored(end_msg, *self._get_color_args()), end="")
def stop_spinning(self):
def stop(self):
r"""Stops the spinner"""
self._is_spinning = False
super().terminate()
time.sleep(0.1)
def exit(self):

View File

@@ -10,9 +10,9 @@ from tabulate import tabulate
from termcolor import colored
from cpl.core.console.background_color_enum import BackgroundColorEnum
from cpl.core.console.console_call import ConsoleCall
from cpl.core.console._call import ConsoleCall
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
from cpl.core.console.spinner_thread import SpinnerThread
from cpl.core.console._spinner import Spinner
class Console:
@@ -464,7 +464,7 @@ class Console:
cls.set_hold_back(True)
spinner = None
if not cls._disabled:
spinner = SpinnerThread(len(message), spinner_foreground_color, spinner_background_color)
spinner = Spinner(len(message), spinner_foreground_color, spinner_background_color)
spinner.start()
return_value = None
@@ -476,7 +476,7 @@ class Console:
cls.close()
if spinner is not None:
spinner.stop_spinning()
spinner.stop()
cls.set_hold_back(False)
cls.set_foreground_color(ForegroundColorEnum.default)

View File

@@ -1,3 +0,0 @@
from .database_settings_name_enum import DatabaseSettingsNameEnum
from .database_settings import DatabaseSettings
from .table_abc import TableABC

View File

@@ -1,18 +0,0 @@
from cpl.core.dependency_injection.scope import Scope
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
class ScopeBuilder:
r"""Class to build :class:`cpl.core.dependency_injection.scope.Scope`"""
def __init__(self, service_provider: ServiceProviderABC) -> None:
self._service_provider = service_provider
def build(self) -> ScopeABC:
r"""Returns scope
Returns:
Object of type :class:`cpl.core.dependency_injection.scope.Scope`
"""
return Scope(self._service_provider)

View File

@@ -1,90 +0,0 @@
from abc import abstractmethod, ABC
from typing import Type
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.typing import T, Source
class ServiceCollectionABC(ABC):
r"""ABC for the class :class:`cpl.core.dependency_injection.service_collection.ServiceCollection`"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def add_db_context(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
r"""Adds database context
Parameter:
db_context: Type[:class:`cpl.core.database.context.database_context_abc.DatabaseContextABC`]
Database context
"""
@abstractmethod
def add_logging(self):
r"""Adds the CPL internal logger"""
@abstractmethod
def add_pipes(self):
r"""Adds the CPL internal pipes as transient"""
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
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
def add_scoped(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with scoped lifetime
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
def add_singleton(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with singleton lifetime
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
def build_service_provider(self) -> ServiceProviderABC:
r"""Creates instance of the service provider
Returns:
Object of type :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
"""

View File

@@ -1,3 +1,2 @@
from .application_environment_abc import ApplicationEnvironmentABC
from .environment_name_enum import EnvironmentNameEnum
from .application_environment import ApplicationEnvironment
from .environment_enum import EnvironmentEnum
from .environment import Environment

View File

@@ -1,95 +0,0 @@
import os
from datetime import datetime
from socket import gethostname
from typing import Optional
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.environment.environment_name_enum import EnvironmentNameEnum
class ApplicationEnvironment(ApplicationEnvironmentABC):
r"""Represents environment of the application
Parameter:
name: :class:`cpl.core.environment.environment_name_enum.EnvironmentNameEnum`
"""
def __init__(self, name: EnvironmentNameEnum = EnvironmentNameEnum.production):
ApplicationEnvironmentABC.__init__(self)
self._environment_name: Optional[EnvironmentNameEnum] = name
self._app_name: Optional[str] = None
self._customer: Optional[str] = None
self._start_time: datetime = datetime.now()
self._end_time: datetime = datetime.now()
self._runtime_directory = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
self._working_directory = os.getcwd()
@property
def environment_name(self) -> str:
return str(self._environment_name.value)
@environment_name.setter
def environment_name(self, environment_name: str):
self._environment_name = EnvironmentNameEnum(environment_name)
@property
def application_name(self) -> str:
return self._app_name if self._app_name is not None else ""
@application_name.setter
def application_name(self, application_name: str):
self._app_name = application_name
@property
def customer(self) -> str:
return self._customer if self._customer is not None else ""
@customer.setter
def customer(self, customer: str):
self._customer = customer
@property
def host_name(self):
return gethostname()
@property
def start_time(self) -> datetime:
return self._start_time
@property
def end_time(self) -> datetime:
return self._end_time
@end_time.setter
def end_time(self, end_time: datetime):
self._end_time = end_time
@property
def date_time_now(self) -> datetime:
return datetime.now()
@property
def working_directory(self) -> str:
return str(self._working_directory)
@property
def runtime_directory(self) -> str:
return str(self._runtime_directory)
def set_runtime_directory(self, runtime_directory: str):
if runtime_directory != "":
self._runtime_directory = runtime_directory
return
self._runtime_directory = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def set_working_directory(self, working_directory: str):
if working_directory != "":
self._working_directory = working_directory
os.chdir(self._working_directory)
return
self._working_directory = os.path.abspath("./")
os.chdir(self._working_directory)

View File

@@ -1,98 +0,0 @@
from abc import ABC, abstractmethod
from datetime import datetime
class ApplicationEnvironmentABC(ABC):
r"""ABC of the class :class:`cpl.core.environment.application_environment.ApplicationEnvironment`"""
@abstractmethod
def __init__(self):
pass
@property
@abstractmethod
def environment_name(self) -> str:
pass
@environment_name.setter
@abstractmethod
def environment_name(self, environment_name: str):
pass
@property
@abstractmethod
def application_name(self) -> str:
pass
@application_name.setter
@abstractmethod
def application_name(self, application_name: str):
pass
@property
@abstractmethod
def customer(self) -> str:
pass
@customer.setter
@abstractmethod
def customer(self, customer: str):
pass
@property
@abstractmethod
def host_name(self) -> str:
pass
@property
@abstractmethod
def start_time(self) -> datetime:
pass
@start_time.setter
@abstractmethod
def start_time(self, start_time: datetime):
pass
@property
@abstractmethod
def end_time(self):
pass
@end_time.setter
@abstractmethod
def end_time(self, end_time: datetime):
pass
@property
@abstractmethod
def date_time_now(self) -> datetime:
pass
@property
@abstractmethod
def working_directory(self) -> str:
pass
@property
@abstractmethod
def runtime_directory(self) -> str:
pass
@abstractmethod
def set_runtime_directory(self, runtime_directory: str):
r"""Sets the current runtime directory
Parameter:
runtime_directory: :class:`str`
Path of the runtime directory
"""
@abstractmethod
def set_working_directory(self, working_directory: str):
r"""Sets the current working directory
Parameter:
working_directory: :class:`str`
Path of the current working directory
"""

View File

@@ -1,22 +1,58 @@
import os
from socket import gethostname
from typing import Optional, Type
from cpl.core.environment.environment_enum import EnvironmentEnum
from cpl.core.typing import T
from cpl.core.utils.get_value import get_value
class Environment:
_environment = "production"
r"""Represents environment of the application
Parameter:
name: :class:`cpl.core.environment.environment_name_enum.EnvironmentNameEnum`
"""
@classmethod
def get_environment(cls):
return cls._environment
return cls.get("ENVIRONMENT", str, EnvironmentEnum.production.value)
@classmethod
def set_environment(cls, environment: str):
if environment not in ["development", "staging", "production"]:
raise ValueError("Invalid environment")
Environment._environment = environment
assert environment is not None and environment != "", "environment must not be None or empty"
assert environment.lower() in [
e.value for e in EnvironmentEnum
], f"environment must be one of {[e.value for e in EnvironmentEnum]}"
cls.set("ENVIRONMENT", environment.lower())
@classmethod
def get_app_name(cls) -> str:
return cls.get("APP_NAME", str)
@classmethod
def set_app_name(cls, app_name: str):
cls.set("APP_NAME", app_name)
@staticmethod
def get_host_name() -> str:
return gethostname()
@staticmethod
def get_cwd() -> str:
return os.getcwd()
@staticmethod
def set_cwd(working_directory: str):
assert working_directory is not None and working_directory != "", "working_directory must not be None or empty"
os.chdir(working_directory)
@staticmethod
def set(key: str, value: T):
assert key is not None and key != "", "key must not be None or empty"
os.environ[key] = str(value)
@staticmethod
def get(key: str, cast_type: Type[T], default: Optional[T] = None) -> Optional[T]:

View File

@@ -1,7 +1,7 @@
from enum import Enum
class EnvironmentNameEnum(Enum):
class EnvironmentEnum(Enum):
production = "production"
staging = "staging"
testing = "testing"

View File

@@ -1,94 +0,0 @@
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()

View File

@@ -3,7 +3,6 @@ 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
@@ -13,6 +12,16 @@ class Logger(LoggerABC):
_level = LogLevelEnum.info
_levels = [x for x in LogLevelEnum]
# 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, source: Source, file_prefix: str = None):
LoggerABC.__init__(self)
assert source is not None and source != "", "Source cannot be None or empty"
@@ -22,7 +31,18 @@ class Logger(LoggerABC):
file_prefix = "app"
self._file_prefix = file_prefix
self._writer = LogWriter.get_instance(self._file_prefix)
self._create_log_dir()
@property
def log_file(self):
return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
@staticmethod
def _create_log_dir():
if os.path.exists("logs"):
return
os.makedirs("logs")
@classmethod
def set_level(cls, level: LogLevelEnum):
@@ -31,6 +51,24 @@ class Logger(LoggerABC):
else:
raise ValueError(f"Invalid log level: {level}")
@staticmethod
def _ensure_file_size(log_file: str):
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):
file = self.log_file
self._ensure_file_size(file)
with open(file, "a") as log_file:
log_file.write(content + "\n")
log_file.close()
def _log(self, level: LogLevelEnum, *messages: Messages):
try:
if self._levels.index(level) < self._levels.index(self._level):
@@ -39,7 +77,8 @@ class Logger(LoggerABC):
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)
self._write_log_to_file(formatted_message)
Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{formatted_message}\033[0m")
except Exception as e:
print(f"Error while logging: {e} -> {traceback.format_exc()}")

View File

@@ -1,9 +1,13 @@
from cpl.core.pipes.pipe_abc import PipeABC
from cpl.core.typing import T
class BoolPipe(PipeABC):
def __init__(self):
pass
class BoolPipe[bool](PipeABC):
def transform(self, value: bool, *args):
return "True" if value else "False"
@staticmethod
def to_str(value: T, *args):
return str(value).lower()
@staticmethod
def from_str(value: str, *args) -> T:
return value in ("True", "true", "1", "yes", "y", "Y")

View File

@@ -1,20 +1,19 @@
from cpl.core.pipes.pipe_abc import PipeABC
from cpl.core.typing import T
class IPAddressPipe(PipeABC):
def __init__(self):
pass
def transform(self, value: list[int], *args):
class IPAddressPipe[list](PipeABC):
@staticmethod
def to_str(value: T, *args) -> str:
string = ""
if len(value) != 4:
raise Exception("Invalid IP")
raise ValueError("Invalid IP")
for i in range(0, len(value)):
byte = value[i]
if byte > 255 or byte < 0:
raise Exception("Invalid IP")
if not 0 <= byte <= 255:
raise ValueError("Invalid IP")
if i == len(value) - 1:
string += f"{byte}"
@@ -22,3 +21,18 @@ class IPAddressPipe(PipeABC):
string += f"{byte}."
return string
@staticmethod
def from_str(value: str, *args) -> T:
parts = value.split(".")
if len(parts) != 4:
raise Exception("Invalid IP")
result = []
for part in parts:
byte = int(part)
if not 0 <= byte <= 255:
raise Exception("Invalid IP")
result.append(byte)
return result

View File

@@ -1,11 +1,16 @@
from abc import ABC, abstractmethod
from typing import Generic
from cpl.core.typing import T
class PipeABC(ABC):
class PipeABC(ABC, Generic[T]):
@staticmethod
@abstractmethod
def __init__(self):
def to_str(value: T, *args) -> str:
pass
@staticmethod
@abstractmethod
def transform(self, value: any, *args):
def from_str(value: str, *args) -> T:
pass

View File

@@ -1,17 +0,0 @@
from cpl_cli.configuration import VersionSettingsNameEnum
from cpl.core.pipes.pipe_abc import PipeABC
class VersionPipe(PipeABC):
def __init__(self):
pass
def transform(self, value: dict, *args):
for atr in VersionSettingsNameEnum:
if atr.value not in value:
raise KeyError(atr.value)
v_str = f"{value[VersionSettingsNameEnum.major.value]}.{value[VersionSettingsNameEnum.minor.value]}"
if value[VersionSettingsNameEnum.micro.value] is not None:
v_str += f".{value[VersionSettingsNameEnum.micro.value]}"
return v_str

View File

@@ -1,6 +1,7 @@
from typing import TypeVar, Any
T = TypeVar("T")
D = TypeVar("D")
R = TypeVar("R")
Service = TypeVar("Service")

View File

@@ -1,3 +1,5 @@
from .b64 import B64
from .credential_manager import CredentialManager
from .string import String
from .json_processor import JSONProcessor
from .pip import Pip
from .string import String

View File

@@ -0,0 +1,43 @@
import base64
from typing import Union
class B64:
@staticmethod
def encode(string: str) -> str:
"""
Encode a string with base64
:param string:
:return:
"""
return base64.b64encode(string.encode("utf-8")).decode("utf-8")
@staticmethod
def decode(string: str) -> str:
"""
Decode a string with base64
:param string:
:return:
"""
return base64.b64decode(string).decode("utf-8")
@staticmethod
def is_b64(sb: Union[str, bytes]) -> bool:
"""
Check if a string is base64 encoded
:param Union[str, bytes] sb:
:return:
:rtype: bool
"""
try:
if isinstance(sb, str):
# If there's any unicode here, an exception will be thrown and the function will return false
sb_bytes = bytes(sb, "ascii")
elif isinstance(sb, bytes):
sb_bytes = sb
else:
raise ValueError("Argument must be string or bytes")
return base64.b64encode(base64.b64decode(sb_bytes)) == sb_bytes
except ValueError:
return False

View File

@@ -40,16 +40,9 @@ def get_value(
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 (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]

View File

@@ -1,7 +1,7 @@
import enum
from inspect import signature, Parameter
from cpl.core.utils import String
from cpl.core.utils.string import String
class JSONProcessor:
@@ -16,7 +16,7 @@ class JSONProcessor:
if parameter.name == "self" or parameter.annotation == Parameter.empty:
continue
name = String.first_to_upper(String.convert_to_camel_case(parameter.name))
name = String.first_to_upper(String.to_camel_case(parameter.name))
name_first_lower = String.first_to_lower(name)
if name in values or name_first_lower in values or name.upper() in values:
value = ""

View File

@@ -1,13 +1,13 @@
import random
import re
import string
import random
class String:
r"""Useful functions for strings"""
@staticmethod
def convert_to_camel_case(chars: str) -> str:
def to_camel_case(s: str) -> str:
r"""Converts string to camel case
Parameter:
@@ -17,16 +17,10 @@ class String:
Returns:
String converted to CamelCase
"""
converted_name = chars
char_set = string.punctuation + " "
for char in char_set:
if char in converted_name:
converted_name = "".join(word.title() for word in converted_name.split(char))
return converted_name
return re.sub(r"(?<!^)(?=[A-Z])", "_", s).lower()
@staticmethod
def convert_to_snake_case(chars: str) -> str:
def to_snake_case(chars: str) -> str:
r"""Converts string to snake case
Parameter:
@@ -56,7 +50,7 @@ class String:
return re.sub(pattern2, r"\1_\2", file_name).lower()
@staticmethod
def first_to_upper(chars: str) -> str:
def first_to_upper(s: str) -> str:
r"""Converts first char to upper
Parameter:
@@ -66,10 +60,10 @@ class String:
Returns:
String with first char as upper
"""
return f"{chars[0].upper()}{chars[1:]}"
return s[0].upper() + s[1:] if s else s
@staticmethod
def first_to_lower(chars: str) -> str:
def first_to_lower(s: str) -> str:
r"""Converts first char to lower
Parameter:
@@ -79,14 +73,24 @@ class String:
Returns:
String with first char as lower
"""
return f"{chars[0].lower()}{chars[1:]}"
return s[0].lower() + s[1:] if s else s
@staticmethod
def random_string(chars: str, length: int) -> str:
def random(length: int, letters=True, digits=False, special_characters=False) -> str:
r"""Creates random string by given chars and length
Returns:
String of random chars
"""
return "".join(random.choice(chars) for _ in range(length))
characters = []
if letters:
characters.append(string.ascii_letters)
if digits:
characters.append(string.digits)
if special_characters:
characters.append(string.punctuation)
return "".join(random.choice(characters) for _ in range(length)) if characters else ""

View File

@@ -3,3 +3,4 @@ colorama==0.4.6
tabulate==0.9.0
termcolor==3.1.0
mysql-connector-python==9.4.0
pynput==1.8.1

View File

@@ -0,0 +1,23 @@
from cpl.dependency import ServiceCollection as _ServiceCollection
from . import mysql
from .database_settings import DatabaseSettings
from .database_settings_name_enum import DatabaseSettingsNameEnum
from .mysql.context import DatabaseContextABC, DatabaseContext
from .table_abc import TableABC
def add_mysql(collection: _ServiceCollection):
from cpl.core.console import Console
from cpl.core.configuration import Configuration
try:
collection.add_singleton(DatabaseContextABC, DatabaseContext)
database_context = collection.build_service_provider().get_service(DatabaseContextABC)
db_settings: DatabaseSettings = Configuration.get(DatabaseSettings)
database_context.connect(db_settings)
except ImportError as e:
Console.error("cpl-translation is not installed", str(e))
_ServiceCollection.with_module(add_mysql, mysql.__name__)

View File

@@ -5,4 +5,4 @@ from cpl.core.typing import Source
class DBLogger(Logger):
def __init__(self, source: Source):
Logger.__init__(self, source, "db")
Logger.__init__(self, source, "db")

View File

@@ -4,8 +4,8 @@ import mysql.connector as sql
from mysql.connector.abstracts import MySQLConnectionAbstract
from mysql.connector.cursor import MySQLCursorBuffered
from cpl.core.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl.core.database.database_settings import DatabaseSettings
from cpl.database.mysql.connection.database_connection_abc import DatabaseConnectionABC
from cpl.database.database_settings import DatabaseSettings
from cpl.core.utils.credential_manager import CredentialManager

View File

@@ -1,12 +1,12 @@
from abc import ABC, abstractmethod
from cpl.core.database.database_settings import DatabaseSettings
from cpl.database.database_settings import DatabaseSettings
from mysql.connector.abstracts import MySQLConnectionAbstract
from mysql.connector.cursor import MySQLCursorBuffered
class DatabaseConnectionABC(ABC):
r"""ABC for the :class:`cpl.core.database.connection.database_connection.DatabaseConnection`"""
r"""ABC for the :class:`cpl.database.connection.database_connection.DatabaseConnection`"""
@abstractmethod
def __init__(self):

View File

@@ -1,10 +1,10 @@
from typing import Optional
from cpl.core.database.connection.database_connection import DatabaseConnection
from cpl.core.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl.core.database.context.database_context_abc import DatabaseContextABC
from cpl.core.database.database_settings import DatabaseSettings
from cpl.database.mysql.connection.database_connection import DatabaseConnection
from cpl.database.mysql.connection.database_connection_abc import DatabaseConnectionABC
from cpl.database.mysql.context.database_context_abc import DatabaseContextABC
from cpl.database.database_settings import DatabaseSettings
from mysql.connector.cursor import MySQLCursorBuffered
@@ -12,7 +12,7 @@ class DatabaseContext(DatabaseContextABC):
r"""Representation of the database context
Parameter:
database_settings: :class:`cpl.core.database.database_settings.DatabaseSettings`
database_settings: :class:`cpl.database.database_settings.DatabaseSettings`
"""
def __init__(self):

View File

@@ -1,11 +1,11 @@
from abc import ABC, abstractmethod
from cpl.core.database.database_settings import DatabaseSettings
from cpl.database.database_settings import DatabaseSettings
from mysql.connector.cursor import MySQLCursorBuffered
class DatabaseContextABC(ABC):
r"""ABC for the :class:`cpl.core.database.context.database_context.DatabaseContext`"""
r"""ABC for the :class:`cpl.database.context.database_context.DatabaseContext`"""
@abstractmethod
def __init__(self, *args):
@@ -21,7 +21,7 @@ class DatabaseContextABC(ABC):
r"""Connects to a database by connection settings
Parameter:
database_settings :class:`cpl.core.database.database_settings.DatabaseSettings`
database_settings :class:`cpl.database.database_settings.DatabaseSettings`
"""
@abstractmethod

View File

@@ -0,0 +1,30 @@
[build-system]
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
build-backend = "setuptools.build_meta"
[project]
name = "cpl-database"
version = "2024.7.0"
description = "CPL database"
readme ="CPL database package"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
]
keywords = ["cpl", "database", "backend", "shared", "library"]
dynamic = ["dependencies", "optional-dependencies"]
[project.urls]
Homepage = "https://www.sh-edraft.de"
[tool.setuptools.packages.find]
where = ["."]
include = ["cpl*"]
[tool.setuptools.dynamic]
dependencies = { file = ["requirements.txt"] }
optional-dependencies.dev = { file = ["requirements.dev.txt"] }

View File

@@ -0,0 +1 @@
black==25.1.0

View File

@@ -0,0 +1,2 @@
cpl-core
cpl-dependency

View File

@@ -1,7 +1,6 @@
from .scope import Scope
from .scope_abc import ScopeABC
from .service_collection import ServiceCollection
from .service_collection_abc import ServiceCollectionABC
from .service_descriptor import ServiceDescriptor
from .service_lifetime_enum import ServiceLifetimeEnum
from .service_provider import ServiceProvider

View File

@@ -1,5 +1,5 @@
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.dependency.scope_abc import ScopeABC
from cpl.dependency.service_provider_abc import ServiceProviderABC
class Scope(ScopeABC):

View File

@@ -2,7 +2,7 @@ from abc import ABC, abstractmethod
class ScopeABC(ABC):
r"""ABC for the class :class:`cpl.core.dependency_injection.scope.Scope`"""
r"""ABC for the class :class:`cpl.dependency.scope.Scope`"""
def __init__(self):
pass
@@ -13,7 +13,7 @@ class ScopeABC(ABC):
r"""Returns to service provider of scope
Returns:
Object of type :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
Object of type :class:`cpl.dependency.service_provider_abc.ServiceProviderABC`
"""
@abstractmethod

View File

@@ -0,0 +1,18 @@
from cpl.dependency.scope import Scope
from cpl.dependency.scope_abc import ScopeABC
from cpl.dependency.service_provider_abc import ServiceProviderABC
class ScopeBuilder:
r"""Class to build :class:`cpl.dependency.scope.Scope`"""
def __init__(self, service_provider: ServiceProviderABC) -> None:
self._service_provider = service_provider
def build(self) -> ScopeABC:
r"""Returns scope
Returns:
Object of type :class:`cpl.dependency.scope.Scope`
"""
return Scope(self._service_provider)

View File

@@ -1,27 +1,26 @@
from typing import Union, Type, Callable, Optional
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.database.context.database_context_abc import DatabaseContextABC
from cpl.core.database.database_settings import DatabaseSettings
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
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.pipes.pipe_abc import PipeABC
from cpl.core.typing import T, Service
from cpl.dependency.service_descriptor import ServiceDescriptor
from cpl.dependency.service_lifetime_enum import ServiceLifetimeEnum
from cpl.dependency.service_provider import ServiceProvider
from cpl.dependency.service_provider_abc import ServiceProviderABC
class ServiceCollection(ServiceCollectionABC):
class ServiceCollection:
r"""Representation of the collection of services"""
def __init__(self, config: ConfigurationABC):
ServiceCollectionABC.__init__(self)
self._configuration: ConfigurationABC = config
_modules: dict[str, Callable] = {}
self._database_context: Optional[DatabaseContextABC] = None
@classmethod
def with_module(cls, func: Callable, name: str = None):
cls._modules[func.__name__ if name is None else name] = func
return cls
def __init__(self):
self._service_descriptors: list[ServiceDescriptor] = []
def _add_descriptor(self, service: Union[type, object], lifetime: ServiceLifetimeEnum, base_type: Callable = None):
@@ -47,10 +46,19 @@ class ServiceCollection(ServiceCollectionABC):
return self
def add_db_context(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
self.add_singleton(DatabaseContextABC, db_context_type)
self._database_context = self.build_service_provider().get_service(DatabaseContextABC)
self._database_context.connect(db_settings)
def add_module(self, module: str | object):
if not isinstance(module, str):
module = module.__name__
if module not in self._modules:
raise ValueError(f"Module {module} not found")
self._modules[module](self)
# def add_mysql(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
# self.add_singleton(DatabaseContextABC, db_context_type)
# self._database_context = self.build_service_provider().get_service(DatabaseContextABC)
# self._database_context.connect(db_settings)
def add_logging(self):
self.add_transient(LoggerABC, Logger)
@@ -74,6 +82,6 @@ class ServiceCollection(ServiceCollectionABC):
return self
def build_service_provider(self) -> ServiceProviderABC:
sp = ServiceProvider(self._service_descriptors, self._configuration, self._database_context)
sp = ServiceProvider(self._service_descriptors)
ServiceProviderABC.set_global_provider(sp)
return sp

View File

@@ -1,6 +1,6 @@
from typing import Union, Optional
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.dependency.service_lifetime_enum import ServiceLifetimeEnum
class ServiceDescriptor:
@@ -9,7 +9,7 @@ class ServiceDescriptor:
Parameter:
implementation: Union[:class:`type`, Optional[:class:`object`]]
Object or type of service
lifetime: :class:`cpl.core.dependency_injection.service_lifetime_enum.ServiceLifetimeEnum`
lifetime: :class:`cpl.dependency.service_lifetime_enum.ServiceLifetimeEnum`
Lifetime of the service
"""

View File

@@ -3,16 +3,15 @@ import typing
from inspect import signature, Parameter, Signature
from typing import Optional
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.configuration import Configuration
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.core.database.context.database_context_abc import DatabaseContextABC
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.core.dependency_injection.scope_builder import ScopeBuilder
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.environment import Environment
from cpl.core.typing import T, R, Source
from cpl.dependency.scope_abc import ScopeABC
from cpl.dependency.scope_builder import ScopeBuilder
from cpl.dependency.service_descriptor import ServiceDescriptor
from cpl.dependency.service_lifetime_enum import ServiceLifetimeEnum
from cpl.dependency.service_provider_abc import ServiceProviderABC
class ServiceProvider(ServiceProviderABC):
@@ -20,25 +19,21 @@ class ServiceProvider(ServiceProviderABC):
Parameter
---------
service_descriptors: list[:class:`cpl.core.dependency_injection.service_descriptor.ServiceDescriptor`]
service_descriptors: list[:class:`cpl.dependency.service_descriptor.ServiceDescriptor`]
Descriptor of the service
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
CPL Configuration
db_context: Optional[:class:`cpl.core.database.context.database_context_abc.DatabaseContextABC`]
db_context: Optional[:class:`cpl.database.context.database_context_abc.DatabaseContextABC`]
Database representation
"""
def __init__(
self,
service_descriptors: list[ServiceDescriptor],
config: ConfigurationABC,
db_context: Optional[DatabaseContextABC],
):
ServiceProviderABC.__init__(self)
self._service_descriptors: list[ServiceDescriptor] = service_descriptors
self._configuration: ConfigurationABC = config
self._database_context = db_context
self._scope: Optional[ScopeABC] = None
def _find_service(self, service_type: type) -> Optional[ServiceDescriptor]:
@@ -94,17 +89,14 @@ class ServiceProvider(ServiceProviderABC):
elif issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
elif issubclass(parameter.annotation, ApplicationEnvironmentABC):
params.append(self._configuration.environment)
elif issubclass(parameter.annotation, DatabaseContextABC):
params.append(self._database_context)
elif issubclass(parameter.annotation, Environment):
params.append(Environment)
elif issubclass(parameter.annotation, ConfigurationModelABC):
params.append(self._configuration.get_configuration(parameter.annotation))
params.append(Configuration.get(parameter.annotation))
elif issubclass(parameter.annotation, ConfigurationABC):
params.append(self._configuration)
elif issubclass(parameter.annotation, Configuration):
params.append(Configuration)
else:
params.append(self._get_service(parameter, origin_service_type))
@@ -115,7 +107,6 @@ class ServiceProvider(ServiceProviderABC):
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:
@@ -142,7 +133,7 @@ class ServiceProvider(ServiceProviderABC):
else:
descriptors.append(copy.deepcopy(descriptor))
sb = ScopeBuilder(ServiceProvider(descriptors, self._configuration, self._database_context))
sb = ScopeBuilder(ServiceProvider(descriptors, self._database_context))
return sb.build()
def get_service(self, service_type: T, *args, **kwargs) -> Optional[R]:

View File

@@ -3,12 +3,12 @@ from abc import abstractmethod, ABC
from inspect import Signature, signature
from typing import Optional
from cpl.core.dependency_injection.scope_abc import ScopeABC
from cpl.dependency.scope_abc import ScopeABC
from cpl.core.typing import T, R
class ServiceProviderABC(ABC):
r"""ABC for the class :class:`cpl.core.dependency_injection.service_provider.ServiceProvider`"""
r"""ABC for the class :class:`cpl.dependency.service_provider.ServiceProvider`"""
_provider: Optional["ServiceProviderABC"] = None
@@ -44,7 +44,7 @@ class ServiceProviderABC(ABC):
Parameter
---------
Object of type :class:`cpl.core.dependency_injection.scope_abc.ScopeABC`
Object of type :class:`cpl.dependency.scope_abc.ScopeABC`
Service scope
"""
@@ -54,7 +54,7 @@ class ServiceProviderABC(ABC):
Returns
-------
Object of type :class:`cpl.core.dependency_injection.scope_abc.ScopeABC`
Object of type :class:`cpl.dependency.scope_abc.ScopeABC`
"""
@abstractmethod

View File

@@ -0,0 +1,30 @@
[build-system]
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
build-backend = "setuptools.build_meta"
[project]
name = "cpl-dependency"
version = "2024.7.0"
description = "CPL dependency"
readme ="CPL dependency package"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
]
keywords = ["cpl", "dependency", "backend", "shared", "library"]
dynamic = ["dependencies", "optional-dependencies"]
[project.urls]
Homepage = "https://www.sh-edraft.de"
[tool.setuptools.packages.find]
where = ["."]
include = ["cpl*"]
[tool.setuptools.dynamic]
dependencies = { file = ["requirements.txt"] }
optional-dependencies.dev = { file = ["requirements.dev.txt"] }

View File

@@ -0,0 +1 @@
black==25.1.0

View File

@@ -0,0 +1 @@
cpl-core

View File

@@ -1,3 +1,4 @@
from cpl.dependency import ServiceCollection as _ServiceCollection
from .abc.email_client_abc import EMailClientABC
from .email_client import EMailClient
from .email_client_settings import EMailClientSettings
@@ -6,21 +7,15 @@ from .email_model import EMail
from .mail_logger import MailLogger
def add_mail(self):
def add_mail(collection: _ServiceCollection):
from cpl.core.console import Console
from cpl.core.log import LoggerABC
try:
self.add_singleton(EMailClientABC, EMailClient)
self.add_transient(LoggerABC, MailLogger)
collection.add_singleton(EMailClientABC, EMailClient)
collection.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()
_ServiceCollection.with_module(add_mail, __name__)

View File

@@ -2,7 +2,6 @@ import ssl
from smtplib import SMTP
from typing import Optional
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.utils.credential_manager import CredentialManager
from cpl.mail.abc.email_client_abc import EMailClientABC
from cpl.mail.email_client_settings import EMailClientSettings
@@ -22,10 +21,11 @@ class EMailClient(EMailClientABC):
Settings for mailing
"""
def __init__(self, environment: ApplicationEnvironmentABC, logger: MailLogger, mail_settings: EMailClientSettings):
def __init__(self, logger: MailLogger, mail_settings: EMailClientSettings):
EMailClientABC.__init__(self)
self._environment = environment
assert mail_settings is not None, "mail_settings must not be None"
self._mail_settings = mail_settings
self._logger = logger

View File

@@ -1,10 +1,11 @@
from cpl.dependency import ServiceCollection as _ServiceCollection
from .translate_pipe import TranslatePipe
from .translation_service import TranslationService
from .translation_service_abc import TranslationServiceABC
from .translation_settings import TranslationSettings
def add_translation(self):
def add_translation(collection: _ServiceCollection):
from cpl.core.console import Console
from cpl.core.pipes import PipeABC
from cpl.translation.translate_pipe import TranslatePipe
@@ -12,16 +13,10 @@ def add_translation(self):
from cpl.translation.translation_service_abc import TranslationServiceABC
try:
self.add_singleton(TranslationServiceABC, TranslationService)
self.add_transient(PipeABC, TranslatePipe)
collection.add_singleton(TranslationServiceABC, TranslationService)
collection.add_transient(PipeABC, TranslatePipe)
except ImportError as e:
Console.error("cpl-translation is not installed", str(e))
def init():
from cpl.core.dependency_injection import ServiceCollection
ServiceCollection.add_translation = add_translation
init()
_ServiceCollection.with_module(add_translation, __name__)

View File

@@ -1,7 +1,7 @@
from cpl.core.application import ApplicationABC
from cpl.application import ApplicationABC
from cpl.core.configuration import ConfigurationABC
from cpl.core.console import Console
from cpl.core.dependency_injection import ServiceProviderABC
from cpl.dependency import ServiceProviderABC
class Application(ApplicationABC):

View File

@@ -1,5 +1,5 @@
import asyncio
from cpl.core.application import ApplicationBuilder
from cpl.application import ApplicationBuilder
from application import Application
from startup import Startup

View File

@@ -1,7 +1,7 @@
from cpl.core.application.async_startup_abc import AsyncStartupABC
from cpl.application.async_startup_abc import AsyncStartupABC
from cpl.core.configuration import ConfigurationABC
from cpl.core.dependency_injection import ServiceProviderABC, ServiceCollectionABC
from cpl.core.environment import ApplicationEnvironment
from cpl.dependency import ServiceProviderABC, ServiceCollection
from cpl.core.environment import Environment
class Startup(AsyncStartupABC):
@@ -9,11 +9,9 @@ class Startup(AsyncStartupABC):
AsyncStartupABC.__init__(self)
async def configure_configuration(
self, configuration: ConfigurationABC, environment: ApplicationEnvironment
self, configuration: ConfigurationABC, environment: Environment
) -> ConfigurationABC:
return configuration
async def configure_services(
self, services: ServiceCollectionABC, environment: ApplicationEnvironment
) -> ServiceProviderABC:
async def configure_services(self, services: ServiceCollection, environment: Environment) -> ServiceProviderABC:
return services.build_service_provider()

View File

@@ -25,6 +25,7 @@ if __name__ == "__main__":
Console.spinner(
"Test:", test_spinner, spinner_foreground_color=ForegroundColorEnum.cyan, text_foreground_color="green"
)
Console.write_line("HOLD BACK")
# opts = [
# 'Option 1',
# 'Option 2',

View File

@@ -1,17 +1,17 @@
from typing import Optional
from cpl.core.application import ApplicationABC
from cpl.core.configuration import ConfigurationABC
from cpl.application import ApplicationABC
from cpl.core.console import Console
from cpl.core.dependency_injection import ServiceProviderABC
from cpl.core.environment import Environment
from cpl.core.log import LoggerABC
from model.user_repo_abc import UserRepoABC
from cpl.dependency import ServiceProviderABC
from model.user_repo import UserRepo
from model.user_repo_abc import UserRepoABC
class Application(ApplicationABC):
def __init__(self, config: ConfigurationABC, services: ServiceProviderABC):
ApplicationABC.__init__(self, config, services)
def __init__(self, services: ServiceProviderABC):
ApplicationABC.__init__(self, services)
self._logger: Optional[LoggerABC] = None
@@ -19,10 +19,8 @@ class Application(ApplicationABC):
self._logger = self._services.get_service(LoggerABC)
def main(self):
self._logger.header(f"{self._configuration.environment.application_name}:")
self._logger.debug(f"Host: {self._configuration.environment.host_name}")
self._logger.debug(f"Environment: {self._configuration.environment.environment_name}")
self._logger.debug(f"Customer: {self._configuration.environment.customer}")
self._logger.debug(f"Host: {Environment.get_host_name()}")
self._logger.debug(f"Environment: {Environment.get_environment()}")
user_repo: UserRepo = self._services.get_service(UserRepoABC)
if len(user_repo.get_users()) == 0:

View File

@@ -1,4 +1,4 @@
from cpl.core.application import ApplicationBuilder
from cpl.application import ApplicationBuilder
from application import Application
from startup import Startup

View File

@@ -1,4 +1,4 @@
from cpl.core.database import TableABC
from cpl.database import TableABC
class CityModel(TableABC):

View File

@@ -1,7 +0,0 @@
from cpl.core.database import DatabaseSettings
from cpl.core.database.context import DatabaseContext
class DBContext(DatabaseContext):
def __init__(self):
DatabaseContext.__init__(self)

View File

@@ -1,4 +1,4 @@
from cpl.core.database import TableABC
from cpl.database import TableABC
from .city_model import CityModel

View File

@@ -1,5 +1,5 @@
from cpl.core.console import Console
from cpl.core.database.context import DatabaseContextABC
from cpl.database.mysql.context import DatabaseContextABC
from .city_model import CityModel
from .user_model import UserModel

View File

@@ -1,11 +1,9 @@
from cpl.core.application import StartupABC
from cpl.core.configuration import ConfigurationABC
from cpl.core.database import DatabaseSettings
from cpl.core.dependency_injection import ServiceCollectionABC, ServiceProviderABC
from cpl.core.environment import ApplicationEnvironmentABC
from cpl.application import StartupABC
from cpl.core.configuration import Configuration
from cpl.core.environment import Environment
from cpl.core.log import Logger, LoggerABC
from model.db_context import DBContext
from cpl.database import mysql
from cpl.dependency import ServiceCollection
from model.user_repo import UserRepo
from model.user_repo_abc import UserRepoABC
@@ -16,28 +14,15 @@ class Startup(StartupABC):
self._configuration = None
def configure_configuration(
self, configuration: ConfigurationABC, environment: ApplicationEnvironmentABC
) -> ConfigurationABC:
configuration.add_environment_variables("PYTHON_")
configuration.add_environment_variables("CPL_")
def configure_configuration(self, configuration: Configuration, environment: Environment):
configuration.add_json_file(f"appsettings.json")
configuration.add_json_file(f"appsettings.{configuration.environment.environment_name}.json")
configuration.add_json_file(f"appsettings.{configuration.environment.host_name}.json", optional=True)
configuration.add_json_file(f"appsettings.{environment.get_environment()}.json")
configuration.add_json_file(f"appsettings.{environment.get_host_name()}.json", optional=True)
self._configuration = configuration
return configuration
def configure_services(
self, services: ServiceCollectionABC, environment: ApplicationEnvironmentABC
) -> ServiceProviderABC:
# Create and connect to database
self._configuration.parse_console_arguments(services)
db_settings: DatabaseSettings = self._configuration.get_configuration(DatabaseSettings)
services.add_db_context(DBContext, db_settings)
def configure_services(self, services: ServiceCollection, environment: Environment):
services.add_module(mysql)
services.add_singleton(UserRepoABC, UserRepo)
services.add_singleton(LoggerABC, Logger)
return services.build_service_provider()

View File

@@ -1,8 +1,8 @@
from cpl.core.application import ApplicationABC
from cpl.application import ApplicationABC
from cpl.core.configuration import ConfigurationABC
from cpl.core.console.console import Console
from cpl.core.dependency_injection import ServiceProviderABC
from cpl.core.dependency_injection.scope import Scope
from cpl.dependency import ServiceProviderABC
from cpl.dependency.scope import Scope
from di.static_test import StaticTest
from di.test_abc import TestABC
from di.test_service import TestService

View File

@@ -1,4 +1,4 @@
from cpl.core.application import ApplicationBuilder
from cpl.application import ApplicationBuilder
from di.application import Application
from di.startup import Startup

View File

@@ -1,7 +1,7 @@
from cpl.core.application import StartupABC
from cpl.application import StartupABC
from cpl.core.configuration import ConfigurationABC
from cpl.core.dependency_injection import ServiceProviderABC, ServiceCollectionABC
from cpl.core.environment import ApplicationEnvironment
from cpl.dependency import ServiceProviderABC, ServiceCollection
from cpl.core.environment import Environment
from di.test1_service import Test1Service
from di.test2_service import Test2Service
from di.test_abc import TestABC
@@ -14,14 +14,10 @@ class Startup(StartupABC):
def __init__(self):
StartupABC.__init__(self)
def configure_configuration(
self, configuration: ConfigurationABC, environment: ApplicationEnvironment
) -> ConfigurationABC:
def configure_configuration(self, configuration: ConfigurationABC, environment: Environment) -> ConfigurationABC:
return configuration
def configure_services(
self, services: ServiceCollectionABC, environment: ApplicationEnvironment
) -> ServiceProviderABC:
def configure_services(self, services: ServiceCollection, environment: Environment) -> ServiceProviderABC:
services.add_scoped(TestService)
services.add_scoped(DITesterService)

Some files were not shown because too many files have changed in this diff Show More