Compare commits

..

3 Commits

Author SHA1 Message Date
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
190 changed files with 722 additions and 2541 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_abc import ServiceCollectionABC
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: ServiceCollectionABC):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection_abc`
"""

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_abc import ServiceCollectionABC
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: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
services: :class:`cpl.dependency.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

View File

@@ -1,9 +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.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.configuration import Configuration
from cpl.dependency.service_collection_abc import ServiceCollectionABC
from cpl.core.environment import Environment
class StartupABC(ABC):
@@ -14,7 +13,7 @@ class StartupABC(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:
@@ -23,10 +22,10 @@ class StartupABC(ABC):
"""
@abstractmethod
def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
def configure_services(self, service: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
services: :class:`cpl.dependency.service_collection_abc`
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_abc import ServiceCollectionABC
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: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
services: :class:`cpl.dependency.service_collection_abc`
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,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,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,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

@@ -59,9 +59,7 @@ class LogWriter:
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"
)
Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{content}\033[0m")
@property
def log_file(self):

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

@@ -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.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.connection.database_connection import DatabaseConnection
from cpl.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl.database.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

@@ -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

@@ -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,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,13 +1,12 @@
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.database.context.database_context_abc import DatabaseContextABC
from cpl.database.database_settings import DatabaseSettings
from cpl.dependency.service_collection_abc import ServiceCollectionABC
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
from cpl.core.log.logger import Logger
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.pipes.pipe_abc import PipeABC
@@ -17,10 +16,8 @@ from cpl.core.typing import T, Service
class ServiceCollection(ServiceCollectionABC):
r"""Representation of the collection of services"""
def __init__(self, config: ConfigurationABC):
def __init__(self):
ServiceCollectionABC.__init__(self)
self._configuration: ConfigurationABC = config
self._database_context: Optional[DatabaseContextABC] = None
self._service_descriptors: list[ServiceDescriptor] = []
@@ -74,6 +71,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, self._database_context)
ServiceProviderABC.set_global_provider(sp)
return sp

View File

@@ -1,14 +1,14 @@
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.database.context.database_context_abc import DatabaseContextABC
from cpl.database.database_settings import DatabaseSettings
from cpl.dependency.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`"""
r"""ABC for the class :class:`cpl.dependency.service_collection.ServiceCollection`"""
@abstractmethod
def __init__(self):
@@ -19,7 +19,7 @@ class ServiceCollectionABC(ABC):
r"""Adds database context
Parameter:
db_context: Type[:class:`cpl.core.database.context.database_context_abc.DatabaseContextABC`]
db_context: Type[:class:`cpl.database.context.database_context_abc.DatabaseContextABC`]
Database context
"""
@@ -50,7 +50,7 @@ class ServiceCollectionABC(ABC):
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
self: :class:`cpl.dependency.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
@@ -64,7 +64,7 @@ class ServiceCollectionABC(ABC):
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
self: :class:`cpl.dependency.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
@@ -78,7 +78,7 @@ class ServiceCollectionABC(ABC):
Object of the service
Returns:
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
self: :class:`cpl.dependency.service_collection_abc.ServiceCollectionABC`
"""
@abstractmethod
@@ -86,5 +86,5 @@ class ServiceCollectionABC(ABC):
r"""Creates instance of the service provider
Returns:
Object of type :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
Object of type :class:`cpl.dependency.service_provider_abc.ServiceProviderABC`
"""

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,16 @@ 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.database.context.database_context_abc import DatabaseContextABC
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,24 +20,22 @@ 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
@@ -94,17 +92,17 @@ 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, Environment):
params.append(Environment)
elif issubclass(parameter.annotation, DatabaseContextABC):
params.append(self._database_context)
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 +113,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 +139,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

@@ -18,7 +18,7 @@ def add_mail(self):
def init():
from cpl.core.dependency_injection import ServiceCollection
from cpl.dependency import ServiceCollection
ServiceCollection.add_mail = add_mail

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

@@ -19,7 +19,7 @@ def add_translation(self):
def init():
from cpl.core.dependency_injection import ServiceCollection
from cpl.dependency import ServiceCollection
ServiceCollection.add_translation = add_translation

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, ServiceCollectionABC
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: ServiceCollectionABC, environment: Environment) -> ServiceProviderABC:
return services.build_service_provider()

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,5 +1,5 @@
from cpl.core.database import DatabaseSettings
from cpl.core.database.context import DatabaseContext
from cpl.database import DatabaseSettings
from cpl.database.context import DatabaseContext
class DBContext(DatabaseContext):

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.context import DatabaseContextABC
from .city_model import CityModel
from .user_model import UserModel

View File

@@ -1,8 +1,8 @@
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.database import DatabaseSettings
from cpl.dependency import ServiceCollectionABC, ServiceProviderABC
from cpl.core.environment import Environment
from cpl.core.log import Logger, LoggerABC
from model.db_context import DBContext
@@ -16,28 +16,18 @@ 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:
def configure_services(self, services: ServiceCollectionABC, environment: Environment):
# Create and connect to database
self._configuration.parse_console_arguments(services)
db_settings: DatabaseSettings = self._configuration.get_configuration(DatabaseSettings)
db_settings: DatabaseSettings = self._configuration.get(DatabaseSettings)
services.add_db_context(DBContext, db_settings)
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, ServiceCollectionABC
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: ServiceCollectionABC, environment: Environment) -> ServiceProviderABC:
services.add_scoped(TestService)
services.add_scoped(DITesterService)

View File

@@ -1,5 +1,5 @@
from cpl.core.configuration import ConfigurationABC
from cpl.core.dependency_injection import ServiceProvider, ServiceProviderABC
from cpl.dependency import ServiceProvider, ServiceProviderABC
from di.test_service import TestService

View File

@@ -1,21 +1,23 @@
import time
from typing import Optional
from cpl.core.application.application_abc import ApplicationABC
from cpl.core.configuration import ConfigurationABC
from cpl.application.application_abc import ApplicationABC
from cpl.core.configuration import Configuration
from cpl.core.console import Console
from cpl.core.dependency_injection import ServiceProviderABC
from cpl.dependency import ServiceProviderABC
from cpl.core.environment import Environment
from cpl.core.log import LoggerABC
from cpl.core.pipes import IPAddressPipe
from cpl.mail import EMail, EMailClientABC
from test_settings import TestSettings
from test_service import TestService
from cpl.query.extension.list import List
from test_service import TestService
from test_settings import TestSettings
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
self._mailer: Optional[EMailClientABC] = None
@@ -25,7 +27,7 @@ class Application(ApplicationABC):
mail.add_header("Content-Type: text/plain; charset=utf-8")
mail.add_header("Content-Transfer-Encoding: quoted-printable")
mail.add_receiver("sven.heidemann@sh-edraft.de")
mail.subject = f"Test - {self._configuration.environment.host_name}"
mail.subject = f"Test - {Environment.get_host_name()}"
mail.body = "Dies ist ein Test :D"
self._mailer.send_mail(mail)
@@ -38,14 +40,8 @@ class Application(ApplicationABC):
self._mailer = self._services.get_service(EMailClientABC)
def main(self):
self._configuration.parse_console_arguments(self._services)
if self._configuration.environment.application_name != "":
self._logger.header(f"{self._configuration.environment.application_name}:")
self._logger.debug(f"Args: {self._configuration.additional_arguments}")
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()}")
Console.write_line(List(int, range(0, 10)).select(lambda x: f"x={x}").to_list())
Console.spinner("Test", self._wait, 2, spinner_foreground_color="red")
test: TestService = self._services.get_service(TestService)
@@ -61,12 +57,12 @@ class Application(ApplicationABC):
with self._services.create_scope() as s:
Console.write_line("with scope", s)
test_settings = self._configuration.get_configuration(TestSettings)
test_settings = Configuration.get(TestSettings)
Console.write_line(test_settings.value)
Console.write_line("reload config")
self._configuration.add_json_file(f"appsettings.json")
self._configuration.add_json_file(f"appsettings.{self._environment.environment_name}.json")
self._configuration.add_json_file(f"appsettings.{self._environment.host_name}.json", optional=True)
test_settings1 = self._configuration.get_configuration(TestSettings)
Configuration.add_json_file(f"appsettings.json")
Configuration.add_json_file(f"appsettings.{Environment.get_environment()}.json")
Configuration.add_json_file(f"appsettings.{Environment.get_host_name()}.json", optional=True)
test_settings1 = Configuration.get(TestSettings)
Console.write_line(test_settings1.value)
# self.test_send_mail()
# self.test_send_mail()

View File

@@ -1,14 +0,0 @@
from cpl.core.configuration import ConfigurationABC, ArgumentExecutableABC
from cpl.core.console import Console
from cpl.core.environment import ApplicationEnvironmentABC
class GenerateArgument(ArgumentExecutableABC):
def __init__(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
ArgumentExecutableABC.__init__(self)
self._config = config
self._env = env
def execute(self, args: list[str]):
Console.error("Generate:")
Console.write_line(args, self._env.environment_name)

View File

@@ -1,10 +0,0 @@
from cpl.core.configuration import ArgumentExecutableABC
from cpl.core.console import Console
class InstallArgument(ArgumentExecutableABC):
def __init__(self):
ArgumentExecutableABC.__init__(self)
def execute(self, args: list[str]):
Console.write_line("Install:", args)

View File

@@ -1,15 +1,13 @@
from application import Application
from cpl.core.application import ApplicationBuilder
from cpl.application import ApplicationBuilder
from test_extension import TestExtension
from startup import Startup
from test_startup_extension import TestStartupExtension
from parameter_startup import ParameterStartup
def main():
app_builder = ApplicationBuilder(Application)
app_builder.use_startup(Startup)
app_builder.use_extension(ParameterStartup)
app_builder.use_extension(TestStartupExtension)
app_builder.use_extension(TestExtension)
app_builder.build().run()

View File

@@ -1,38 +0,0 @@
from arguments.generate_argument import GenerateArgument
from arguments.install_argument import InstallArgument
from cpl.core.application import StartupExtensionABC
from cpl.core.configuration import ConfigurationABC, ArgumentTypeEnum
from cpl.core.dependency_injection import ServiceCollectionABC
from cpl.core.environment import ApplicationEnvironmentABC
class ParameterStartup(StartupExtensionABC):
def __init__(self):
StartupExtensionABC.__init__(self)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
config.create_console_argument(
ArgumentTypeEnum.Executable, "", "generate", ["g", "G"], GenerateArgument
).add_console_argument(ArgumentTypeEnum.Variable, "", "abc", ["a", "A"], " ").add_console_argument(
ArgumentTypeEnum.Variable, "", "class", ["c", "C"], " "
).add_console_argument(
ArgumentTypeEnum.Variable, "", "enum", ["e", "E"], " "
).add_console_argument(
ArgumentTypeEnum.Variable, "", "service", ["s", "S"], " "
).add_console_argument(
ArgumentTypeEnum.Variable, "", "settings", ["st", "ST"], " "
).add_console_argument(
ArgumentTypeEnum.Variable, "", "thread", ["t", "T"], " "
).add_console_argument(
ArgumentTypeEnum.Variable, "-", "o", ["o", "O"], "="
).add_console_argument(
ArgumentTypeEnum.Flag, "--", "virtual", ["v", "V"]
)
config.create_console_argument(
ArgumentTypeEnum.Executable, "", "install", ["i", "I"], InstallArgument
).add_console_argument(ArgumentTypeEnum.Flag, "--", "virtual", ["v", "V"]).add_console_argument(
ArgumentTypeEnum.Flag, "--", "simulate", ["s", "S"]
)
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
services.add_transient(GenerateArgument).add_singleton(InstallArgument)

View File

@@ -1,9 +1,7 @@
from cpl.core.application import StartupABC
from cpl.core.configuration import ConfigurationABC
from cpl.core.dependency_injection import ServiceCollectionABC, ServiceProviderABC
from cpl.core.environment import ApplicationEnvironmentABC
from cpl.core.log import Logger, LoggerABC
from cpl.mail import EMailClient, EMailClientABC
from cpl.application import StartupABC
from cpl.core.configuration import Configuration
from cpl.dependency import ServiceCollectionABC, ServiceProviderABC
from cpl.core.environment import Environment
from cpl.core.pipes import IPAddressPipe
from test_service import TestService
@@ -12,14 +10,12 @@ class Startup(StartupABC):
def __init__(self):
StartupABC.__init__(self)
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC) -> ConfigurationABC:
config.add_environment_variables("PYTHON_")
config.add_environment_variables("CPLT_")
def configure_configuration(self, config: Configuration, env: Environment):
config.add_json_file(f"appsettings.json")
config.add_json_file(f"appsettings.{config.environment.environment_name}.json")
config.add_json_file(f"appsettings.{config.environment.host_name}.json", optional=True)
config.add_json_file(f"appsettings.{env.get_environment()}.json")
config.add_json_file(f"appsettings.{env.get_host_name()}.json", optional=True)
def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC) -> ServiceProviderABC:
def configure_services(self, services: ServiceCollectionABC, env: Environment):
services.add_logging()
services.add_mail()
services.add_transient(IPAddressPipe)

View File

@@ -1,12 +1,12 @@
from cpl.core.application import ApplicationExtensionABC
from cpl.core.configuration import ConfigurationABC
from cpl.application import ApplicationExtensionABC
from cpl.core.configuration import Configuration
from cpl.core.console import Console
from cpl.core.dependency_injection import ServiceProviderABC
from cpl.dependency import ServiceProviderABC
class TestExtension(ApplicationExtensionABC):
def __init__(self):
ApplicationExtensionABC.__init__(self)
def run(self, config: ConfigurationABC, services: ServiceProviderABC):
def run(self, config: Configuration, services: ServiceProviderABC):
Console.write_line("Hello World from App Extension")

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