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

Updated config & environment

Browse Source
This commit is contained in:
Sven Heidemann
2025-09-16 08:39:00 +02:00
parent 5edabbf8c1
commit 5f25400bcd
134 changed files with 366 additions and 2233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/cpl-core/cpl/core/application/application_abc.py
View File

@@ -1,10 +1,10 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.configuration import Configuration
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
from cpl.core.environment import Environment
class ApplicationABC(ABC):
@@ -18,9 +18,7 @@ class ApplicationABC(ABC):
"""
@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):

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

@@ -10,6 +10,7 @@ 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
from cpl.core.environment import Environment
class ApplicationBuilder(ApplicationBuilderABC):
@@ -25,9 +26,7 @@ class ApplicationBuilder(ApplicationBuilderABC):
self._app = app
self._startup: Optional[StartupABC | AsyncStartupABC] = None
self._configuration = Configuration()
self._environment = self._configuration.environment
self._services = ServiceCollection(self._configuration)
self._services = ServiceCollection()
self._app_extensions: list[Type[ApplicationExtensionABC | AsyncApplicationExtensionABC]] = []
self._startup_extensions: list[Type[StartupExtensionABC | AsyncStartupABC]] = []
@@ -37,11 +36,18 @@ class ApplicationBuilder(ApplicationBuilderABC):
return self
def use_extension(
self, extension: Type[ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC]
self,
extension: Type[
ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC
],
) -> "ApplicationBuilder":
if (issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)) and extension not in self._app_extensions:
if (
issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)
) and extension not in self._app_extensions:
self._app_extensions.append(extension)
elif (issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)) and extension not in self._startup_extensions:
elif (
issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)
) and extension not in self._startup_extensions:
self._startup_extensions.append(extension)
return self
@@ -49,43 +55,43 @@ class ApplicationBuilder(ApplicationBuilderABC):
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)
extension.configure_configuration(Configuration, Environment)
extension.configure_services(self._services, Environment)
if self._startup is not None:
self._startup.configure_configuration(self._configuration, self._environment)
self._startup.configure_services(self._services, self._environment)
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(self._configuration, self._environment)
await extension.configure_services(self._services, self._environment)
await extension.configure_configuration(Configuration, Environment)
await extension.configure_services(self._services, 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)
await self._startup.configure_configuration(Configuration, Environment)
await self._startup.configure_services(self._services, Environment)
def build(self) -> ApplicationABC:
self._build_startup()
config = self._configuration
config = Configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
extension.run(config, services)
return self._app(config, services)
return self._app(services)
async def build_async(self) -> ApplicationABC:
await self._build_async_startup()
config = self._configuration
config = 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)
return self._app(services)

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

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

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

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

15
src/cpl-core/cpl/core/application/async_startup_abc.py
View File

@@ -1,9 +1,6 @@
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):
@@ -14,19 +11,13 @@ class AsyncStartupABC(ABC):
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`
"""
async def configure_configuration(self):
r"""Creates configuration of application"""
@abstractmethod
async def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
async def configure_services(self, service: ServiceCollectionABC):
r"""Creates service provider
Parameter:
services: :class:`cpl.core.dependency_injection.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

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

@@ -1,8 +1,8 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.configuration.configuration import Configuration
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.core.environment.environment import Environment
class AsyncStartupExtensionABC(ABC):
@@ -13,16 +13,16 @@ 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:

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

@@ -1,9 +1,8 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.configuration import Configuration
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.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,7 +22,7 @@ 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:

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

@@ -1,8 +1,10 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration_abc import ConfigurationABC
from cpl.core.configuration import Configuration
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
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,7 +24,7 @@ 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:

10
src/cpl-core/cpl/core/configuration/__init__.py
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

64
src/cpl-core/cpl/core/configuration/argument_abc.py
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

30
src/cpl-core/cpl/core/configuration/argument_builder.py
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

11
src/cpl-core/cpl/core/configuration/argument_executable_abc.py
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

7
src/cpl-core/cpl/core/configuration/argument_type_enum.py
View File

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

273
src/cpl-core/cpl/core/configuration/configuration.py
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,16 +100,16 @@ 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:
@@ -265,94 +128,18 @@ class Configuration(ConfigurationABC):
else:
configuration = JSONProcessor.process(sub, value)
self.add_configuration(sub, configuration)
cls.set(sub, configuration)
def add_configuration(self, key_type: T, value: any):
self._config[key_type] = value
@classmethod
def set(cls, key: Any, value: T):
if inspect.isclass(key):
key = key.__name__
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
cls._config[key] = value
def for_each_argument(self, call: Callable):
for arg in self._argument_types:
call(arg)
@classmethod
def get(cls, key: Any, default: D = None) -> T | D:
if inspect.isclass(key):
key = key.__name__
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)

140
src/cpl-core/cpl/core/configuration/configuration_abc.py
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.
"""

11
src/cpl-core/cpl/core/configuration/configuration_variable_name_enum.py
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]

40
src/cpl-core/cpl/core/configuration/executable_argument.py
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)

13
src/cpl-core/cpl/core/configuration/flag_argument.py
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)

11
src/cpl-core/cpl/core/configuration/validator_abc.py
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

28
src/cpl-core/cpl/core/configuration/variable_argument.py
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

2
src/cpl-core/cpl/core/database/db_logger.py
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")

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

@@ -1,6 +1,5 @@
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
@@ -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

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

@@ -3,7 +3,7 @@ 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
@@ -11,7 +11,7 @@ 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
@@ -31,13 +31,11 @@ class ServiceProvider(ServiceProviderABC):
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]:

5
src/cpl-core/cpl/core/environment/__init__.py
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

95
src/cpl-core/cpl/core/environment/application_environment.py
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)

98
src/cpl-core/cpl/core/environment/application_environment_abc.py
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
"""

46
src/cpl-core/cpl/core/environment/environment.py
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]:

2
src/cpl-core/cpl/core/environment/environment_name_enum.py → src/cpl-core/cpl/core/environment/environment_enum.py
View File

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

4
src/cpl-core/cpl/core/log/_log_writer.py
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):

14
src/cpl-core/cpl/core/pipes/bool_pipe.py
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")

30
src/cpl-core/cpl/core/pipes/ip_address_pipe.py
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

11
src/cpl-core/cpl/core/pipes/pipe_abc.py
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

17
src/cpl-core/cpl/core/pipes/version_pipe.py
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

1
src/cpl-core/cpl/core/typing.py
View File

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

4
src/cpl-core/cpl/core/utils/__init__.py
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

43
src/cpl-core/cpl/core/utils/b64.py Normal file
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

13
src/cpl-core/cpl/core/utils/get_value.py
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]

4
src/cpl-core/cpl/core/utils/json_processor.py
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 = ""

36
src/cpl-core/cpl/core/utils/string.py
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 ""

1
src/cpl-core/requirements.txt
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

6
src/cpl-mail/cpl/mail/email_client.py
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