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

Updated docs

Browse Source
This commit is contained in:
Sven Heidemann
2023-02-20 15:55:20 +01:00
parent 48d0daabf5
commit 9e28dce5ce
632 changed files with 10917 additions and 6775 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/cpl_core/__init__.py
View File

@@ -1,26 +1,26 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

18
src/cpl_core/application/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.application'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.application"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -28,5 +28,5 @@ from .application_extension_abc import ApplicationExtensionABC
from .startup_abc import StartupABC
from .startup_extension_abc import StartupExtensionABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

5
src/cpl_core/application/application_abc.py
View File

@@ -10,8 +10,7 @@ from cpl_core.environment.application_environment_abc import ApplicationEnvironm
class ApplicationABC(ABC):
r"""ABC for the Application class
Parameters
----------
Parameters:
config: :class:`cpl_core.configuration.configuration_abc.ConfigurationABC`
Contains object loaded from appsettings
services: :class:`cpl_core.dependency_injection.service_provider_abc.ServiceProviderABC`
@@ -34,7 +33,7 @@ class ApplicationABC(ABC):
self.main()
except KeyboardInterrupt:
Console.close()
async def run_async(self):
r"""Entry point

9
src/cpl_core/application/application_builder.py
View File

@@ -12,8 +12,7 @@ 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
---------
Parameter:
app: Type[:class:`cpl_core.application.application_abc.ApplicationABC`]
Application to build
"""
@@ -30,11 +29,13 @@ class ApplicationBuilder(ApplicationBuilderABC):
self._app_extensions: list[Callable] = []
self._startup_extensions: list[Callable] = []
def use_startup(self, startup: Type[StartupABC]) -> 'ApplicationBuilder':
def use_startup(self, startup: Type[StartupABC]) -> "ApplicationBuilder":
self._startup = startup()
return self
def use_extension(self, extension: Type[Union[ApplicationExtensionABC, StartupExtensionABC]]) -> 'ApplicationBuilder':
def use_extension(
self, extension: Type[Union[ApplicationExtensionABC, StartupExtensionABC]]
) -> "ApplicationBuilder":
if issubclass(extension, ApplicationExtensionABC) and extension not in self._app_extensions:
self._app_extensions.append(extension)
elif issubclass(extension, StartupExtensionABC) and extension not in self._startup_extensions:

12
src/cpl_core/application/application_builder_abc.py
View File

@@ -16,8 +16,7 @@ class ApplicationBuilderABC(ABC):
def use_startup(self, startup: Type[StartupABC]):
r"""Sets the custom startup class to use
Parameter
---------
Parameter:
startup: Type[:class:`cpl_core.application.startup_abc.StartupABC`]
Startup class to use
"""
@@ -27,8 +26,7 @@ class ApplicationBuilderABC(ABC):
async def use_startup(self, startup: Type[StartupABC]):
r"""Sets the custom startup class to use async
Parameter
---------
Parameter:
startup: Type[:class:`cpl_core.application.startup_abc.StartupABC`]
Startup class to use
"""
@@ -38,8 +36,7 @@ class ApplicationBuilderABC(ABC):
def build(self) -> ApplicationABC:
r"""Creates custom application object
Returns
-------
Returns:
Object of :class:`cpl_core.application.application_abc.ApplicationABC`
"""
pass
@@ -48,8 +45,7 @@ class ApplicationBuilderABC(ABC):
async def build_async(self) -> ApplicationABC:
r"""Creates custom application object async
Returns
-------
Returns:
Object of :class:`cpl_core.application.application_abc.ApplicationABC`
"""
pass

12
src/cpl_core/application/application_extension_abc.py
View File

@@ -5,12 +5,14 @@ from cpl_core.dependency_injection import ServiceProviderABC
class ApplicationExtensionABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def __init__(self): pass
def run(self, config: ConfigurationABC, services: ServiceProviderABC):
pass
@abstractmethod
def run(self, config: ConfigurationABC, services: ServiceProviderABC): pass
@abstractmethod
async def run(self, config: ConfigurationABC, services: ServiceProviderABC): pass
async def run(self, config: ConfigurationABC, services: ServiceProviderABC):
pass

15
src/cpl_core/application/startup_abc.py
View File

@@ -10,19 +10,18 @@ class StartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self): pass
def __init__(self):
pass
@abstractmethod
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC) -> ConfigurationABC:
r"""Creates configuration of application
Parameter
---------
Parameter:
config: :class:`cpl_core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl_core.environment.application_environment_abc`
Returns
-------
Returns:
Object of :class:`cpl_core.configuration.configuration_abc.ConfigurationABC`
"""
pass
@@ -31,13 +30,11 @@ class StartupABC(ABC):
def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC) -> ServiceProviderABC:
r"""Creates service provider
Parameter
---------
Parameter:
services: :class:`cpl_core.dependency_injection.service_collection_abc`
env: :class:`cpl_core.environment.application_environment_abc`
Returns
-------
Returns:
Object of :class:`cpl_core.dependency_injection.service_provider_abc.ServiceProviderABC`
"""
pass

9
src/cpl_core/application/startup_extension_abc.py
View File

@@ -9,14 +9,14 @@ class StartupExtensionABC(ABC):
r"""ABC for startup extension classes"""
@abstractmethod
def __init__(self): pass
def __init__(self):
pass
@abstractmethod
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
r"""Creates configuration of application
Parameter
---------
Parameter:
config: :class:`cpl_core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl_core.environment.application_environment_abc`
"""
@@ -26,8 +26,7 @@ class StartupExtensionABC(ABC):
def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
r"""Creates service provider
Parameter
---------
Parameter:
services: :class:`cpl_core.dependency_injection.service_collection_abc`
env: :class:`cpl_core.environment.application_environment_abc`
"""

18
src/cpl_core/configuration/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.configuration'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.configuration"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -34,5 +34,5 @@ from .flag_argument import FlagArgument
from .validator_abc import ValidatorABC
from .variable_argument import VariableArgument
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

30
src/cpl_core/configuration/argument_abc.py
View File

@@ -4,19 +4,18 @@ 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
):
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
---------
Parameter:
token: :class:`str`
name: :class:`str`
aliases: list[:class:`str`]
@@ -45,22 +44,21 @@ class ArgumentABC(ABC):
return self._prevent_next_executable
@property
def console_arguments(self) -> list['ArgumentABC']:
def console_arguments(self) -> list["ArgumentABC"]:
return self._console_arguments
def add_console_argument(self, arg_type: ArgumentTypeEnum, *args, **kwargs) -> 'ArgumentABC':
def add_console_argument(self, arg_type: ArgumentTypeEnum, *args, **kwargs) -> "ArgumentABC":
r"""Creates and adds a console argument to known console arguments
Parameter
---------
Parameter:
arg_type: :class:`str`
Specifies the specific type of the argument
Returns
------
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

8
src/cpl_core/configuration/argument_builder.py
View File

@@ -8,10 +8,10 @@ from cpl_core.console import Console
class ArgumentBuilder:
@staticmethod
def build_argument(arg_type: ArgumentTypeEnum, *args, **kwargs) -> Union[
ExecutableArgument, FlagArgument, VariableArgument]:
def build_argument(
arg_type: ArgumentTypeEnum, *args, **kwargs
) -> Union[ExecutableArgument, FlagArgument, VariableArgument]:
argument = None
try:
match arg_type:
@@ -22,7 +22,7 @@ class ArgumentBuilder:
case ArgumentTypeEnum.Variable:
argument = VariableArgument(*args, **kwargs)
case _:
Console.error('Invalid argument type')
Console.error("Invalid argument type")
Console.close()
except TypeError as e:
Console.error(str(e))

9
src/cpl_core/configuration/argument_executable_abc.py
View File

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

1
src/cpl_core/configuration/argument_type_enum.py
View File

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

113
src/cpl_core/configuration/configuration.py
View File

@@ -11,7 +11,9 @@ from cpl_core.configuration.argument_executable_abc import ArgumentExecutableABC
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.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.validator_abc import ValidatorABC
@@ -26,7 +28,6 @@ from cpl_core.type import T
class Configuration(ConfigurationABC):
def __init__(self):
r"""Representation of configuration"""
ConfigurationABC.__init__(self)
@@ -65,52 +66,48 @@ class Configuration(ConfigurationABC):
def _print_info(name: str, message: str):
r"""Prints an info message
Parameter
---------
Parameter:
name: :class:`str`
Info name
message: :class:`str`
Info message
"""
Console.set_foreground_color(ForegroundColorEnum.green)
Console.write_line(f'[{name}] {message}')
Console.write_line(f"[{name}] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
@staticmethod
def _print_warn(name: str, message: str):
r"""Prints a warning
Parameter
---------
Parameter:
name: :class:`str`
Warning name
message: :class:`str`
Warning message
"""
Console.set_foreground_color(ForegroundColorEnum.yellow)
Console.write_line(f'[{name}] {message}')
Console.write_line(f"[{name}] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
@staticmethod
def _print_error(name: str, message: str):
r"""Prints an error
Parameter
---------
Parameter:
name: :class:`str`
Error name
message: :class:`str`
Error message
"""
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f'[{name}] {message}')
Console.write_line(f"[{name}] {message}")
Console.set_foreground_color(ForegroundColorEnum.default)
def _set_variable(self, name: str, value: any):
r"""Sets variable to given value
Parameter
---------
Parameter:
name: :class:`str`
Name of the variable
value: :class:`any`
@@ -131,34 +128,33 @@ class Configuration(ConfigurationABC):
def _load_json_file(self, file: str, output: bool) -> dict:
r"""Reads the json file
Parameter
---------
Parameter:
file: :class:`str`
Name of the file
output: :class:`bool`
Specifies whether an output should take place
Returns
-------
Returns:
Object of :class:`dict`
"""
try:
# open config file, create if not exists
with open(file, encoding='utf-8') as cfg:
with open(file, encoding="utf-8") as cfg:
# load json
json_cfg = json.load(cfg)
if output:
self._print_info(__name__, f'Loaded config file: {file}')
self._print_info(__name__, f"Loaded config file: {file}")
return json_cfg
except Exception as e:
self._print_error(__name__, f'Cannot load config file: {file}! -> {e}')
self._print_error(__name__, f"Cannot load config file: {file}! -> {e}")
return {}
def _handle_pre_or_post_executables(self, pre: bool, argument: ExecutableArgument, services: ServiceProviderABC):
script_type = 'pre-' if pre else 'post-'
script_type = "pre-" if pre else "post-"
from cpl_cli.configuration.workspace_settings import WorkspaceSettings
workspace: Optional[WorkspaceSettings] = self.get_configuration(WorkspaceSettings)
if workspace is None or len(workspace.scripts) == 0:
return
@@ -172,15 +168,21 @@ class Configuration(ConfigurationABC):
continue
from cpl_cli.command.custom_script_service import CustomScriptService
css: CustomScriptService = services.get_service(CustomScriptService)
if css is None:
continue
Console.write_line()
self._set_variable('ACTIVE_EXECUTABLE', script)
self._set_variable("ACTIVE_EXECUTABLE", script)
css.run(self._additional_arguments)
def _parse_arguments(self, executables: list[ArgumentABC], arg_list: list[str], args_types: list[ArgumentABC]):
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)):
@@ -191,10 +193,14 @@ class Configuration(ConfigurationABC):
# 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:
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)
self._parse_arguments(executables, arg_list[i + 1 :], arg.console_arguments)
# variables
elif isinstance(arg, VariableArgument):
@@ -202,24 +208,32 @@ class Configuration(ConfigurationABC):
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 != ' ':
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)
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.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)
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:
@@ -230,7 +244,7 @@ class Configuration(ConfigurationABC):
if not env_var.startswith(prefix):
continue
self._set_variable(env_var.replace(prefix, ''), os.environ[env_var])
self._set_variable(env_var.replace(prefix, ""), os.environ[env_var])
def add_console_argument(self, argument: ArgumentABC):
self._argument_types.append(argument)
@@ -243,27 +257,27 @@ class Configuration(ConfigurationABC):
if path is not None:
path_root = path
if str(path_root).endswith('/') and not name.startswith('/'):
file_path = f'{path_root}{name}'
if str(path_root).endswith("/") and not name.startswith("/"):
file_path = f"{path_root}{name}"
else:
file_path = f'{path_root}/{name}'
file_path = f"{path_root}/{name}"
if not os.path.isfile(file_path):
if optional is not True:
if output:
self._print_error(__name__, f'File not found: {file_path}')
self._print_error(__name__, f"File not found: {file_path}")
sys.exit()
if output:
self._print_warn(__name__, f'Not Loaded config file: {file_path}')
self._print_warn(__name__, f"Not Loaded config file: {file_path}")
return None
config_from_file = self._load_json_file(file_path, output)
for sub in ConfigurationModelABC.__subclasses__():
for key, value in config_from_file.items():
if sub.__name__ == key or sub.__name__.replace('Settings', '') == key:
if sub.__name__ == key or sub.__name__.replace("Settings", "") == key:
configuration = sub()
configuration.from_dict(value)
self.add_configuration(sub, configuration)
@@ -271,8 +285,15 @@ class Configuration(ConfigurationABC):
def add_configuration(self, key_type: T, value: any):
self._config[key_type] = value
def create_console_argument(self, arg_type: ArgumentTypeEnum, token: str, name: str, aliases: list[str],
*args, **kwargs) -> ArgumentABC:
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
@@ -302,7 +323,7 @@ class Configuration(ConfigurationABC):
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()], '='))
self.add_console_argument(VariableArgument("--", str(arg_name).upper(), [str(arg_name).lower()], "="))
success = False
try:
@@ -310,7 +331,7 @@ class Configuration(ConfigurationABC):
executables: list[ExecutableArgument] = []
self._parse_arguments(executables, arg_list, self._argument_types)
except Exception as e:
Console.error('An error occurred while parsing arguments.')
Console.error("An error occurred while parsing arguments.")
sys.exit()
try:
@@ -333,11 +354,11 @@ class Configuration(ConfigurationABC):
cmd: ArgumentExecutableABC = 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')
self._set_variable("ACTIVE_EXECUTABLE", exe.name)
args = self.get_configuration("ARGS")
if args is not None:
for arg in args.split(' '):
if arg == '':
for arg in args.split(" "):
if arg == "":
continue
self._additional_arguments.append(arg)
@@ -346,7 +367,7 @@ class Configuration(ConfigurationABC):
prevent = exe.prevent_next_executable
success = True
except Exception as e:
Console.error('An error occurred while executing arguments.', traceback.format_exc())
Console.error("An error occurred while executing arguments.", traceback.format_exc())
sys.exit()
return success

56
src/cpl_core/configuration/configuration_abc.py
View File

@@ -10,7 +10,6 @@ from cpl_core.type import T
class ConfigurationABC(ABC):
@abstractmethod
def __init__(self):
r"""ABC for the :class:`cpl_core.configuration.configuration.Configuration`"""
@@ -18,30 +17,34 @@ class ConfigurationABC(ABC):
@property
@abstractmethod
def environment(self) -> ApplicationEnvironmentABC: pass
def environment(self) -> ApplicationEnvironmentABC:
pass
@property
@abstractmethod
def additional_arguments(self) -> list[str]: pass
def additional_arguments(self) -> list[str]:
pass
@property
@abstractmethod
def argument_error_function(self) -> Optional[Callable]: pass
def argument_error_function(self) -> Optional[Callable]:
pass
@argument_error_function.setter
@abstractmethod
def argument_error_function(self, argument_error_function: Callable): pass
def argument_error_function(self, argument_error_function: Callable):
pass
@property
@abstractmethod
def arguments(self) -> list[ArgumentABC]: pass
def arguments(self) -> list[ArgumentABC]:
pass
@abstractmethod
def add_environment_variables(self, prefix: str):
r"""Reads the environment variables
Parameter
---------
Parameter:
prefix: :class:`str`
Prefix of the variables
"""
@@ -51,8 +54,7 @@ class ConfigurationABC(ABC):
def add_console_argument(self, argument: ArgumentABC):
r"""Adds console argument to known console arguments
Parameter
---------
Parameter:
argument: :class:`cpl_core.configuration.console_argument.ConsoleArgumentABC`
Specifies the console argument
"""
@@ -62,8 +64,7 @@ class ConfigurationABC(ABC):
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
---------
Parameter:
name: :class:`str`
Name of the file
optional: :class:`str`
@@ -79,8 +80,7 @@ class ConfigurationABC(ABC):
def add_configuration(self, key_type: T, value: any):
r"""Add configuration object
Parameter
---------
Parameter:
key_type: Union[:class:`str`, :class:`type`]
Type of the value
value: any
@@ -89,12 +89,12 @@ class ConfigurationABC(ABC):
pass
@abstractmethod
def create_console_argument(self, arg_type: ArgumentTypeEnum, token: str, name: str, aliases: list[str],
*args, **kwargs) -> ArgumentABC:
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
---------
Parameter:
token: :class:`str`
Specifies optional beginning of argument
name :class:`str`
@@ -108,8 +108,7 @@ class ConfigurationABC(ABC):
runnable: :class:`cpl_core.configuration.console_argument.ConsoleArgumentABC`
Specifies class to run when called if value is not None
Returns
------
Returns:
Object of :class:`cpl_core.configuration.console_argument.ConsoleArgumentABC`
"""
pass
@@ -118,8 +117,7 @@ class ConfigurationABC(ABC):
def for_each_argument(self, call: Callable):
r"""Iterates through all arguments and calls the call function
Parameter
---------
Parameter:
call: :class:`Callable`
Call for each argument
"""
@@ -129,28 +127,24 @@ class ConfigurationABC(ABC):
def get_configuration(self, search_type: Type[T]) -> Optional[T]:
r"""Returns value from configuration by given type
Parameter
---------
Parameter:
search_type: Union[:class:`str`, Type[:class:`cpl_core.configuration.configuration_model_abc.ConfigurationModelABC`]]
Type to search for
Returns
-------
Returns:
Object of Union[:class:`str`, :class:`cpl_core.configuration.configuration_model_abc.ConfigurationModelABC`]
"""
pass
@abstractmethod
def parse_console_arguments(self, services: 'ServiceProviderABC', error: bool = None) -> bool:
def parse_console_arguments(self, services: "ServiceProviderABC", error: bool = None) -> bool:
r"""Reads the console arguments
Parameter
---------
Parameter:
error: :class:`bool`
Defines is invalid argument error will be shown or not
Returns
-------
Returns:
Bool to specify if executables were executed or not.
"""
pass

4
src/cpl_core/configuration/configuration_model_abc.py
View File

@@ -2,7 +2,6 @@ from abc import ABC, abstractmethod
class ConfigurationModelABC(ABC):
@abstractmethod
def __init__(self):
r"""ABC for settings representation"""
@@ -12,8 +11,7 @@ class ConfigurationModelABC(ABC):
def from_dict(self, settings: dict):
r"""Converts attributes to dict
Parameter
---------
Parameter:
settings: :class:`dict`
"""
pass

7
src/cpl_core/configuration/configuration_variable_name_enum.py
View File

@@ -2,10 +2,9 @@ from enum import Enum
class ConfigurationVariableNameEnum(Enum):
environment = 'ENVIRONMENT'
name = 'NAME'
customer = 'CUSTOMER'
environment = "ENVIRONMENT"
name = "NAME"
customer = "CUSTOMER"
@staticmethod
def to_list():

24
src/cpl_core/configuration/executable_argument.py
View File

@@ -6,17 +6,16 @@ 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
):
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
@@ -35,8 +34,7 @@ class ExecutableArgument(ArgumentABC):
return self._validators
def run(self, args: list[str]):
r"""Executes runnable if exists
"""
r"""Executes runnable if exists"""
if self._executable is None:
return
self._executable.execute(args)

17
src/cpl_core/configuration/flag_argument.py
View File

@@ -2,13 +2,12 @@ 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
):
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)

9
src/cpl_core/configuration/validator_abc.py
View File

@@ -2,9 +2,10 @@ from abc import ABC, abstractmethod
class ValidatorABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def __init__(self): pass
@abstractmethod
def validate(self) -> bool: pass
def validate(self) -> bool:
pass

20
src/cpl_core/configuration/variable_argument.py
View File

@@ -2,17 +2,17 @@ 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
):
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 = ''
self._value: str = ""
ArgumentABC.__init__(self, token, name, aliases, prevent_next_executable, console_arguments)

18
src/cpl_core/console/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.console'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.console"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -27,5 +27,5 @@ from .console_call import ConsoleCall
from .foreground_color_enum import ForegroundColorEnum
from .spinner_thread import SpinnerThread
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

19
src/cpl_core/console/background_color_enum.py
View File

@@ -2,13 +2,12 @@ from enum import Enum
class BackgroundColorEnum(Enum):
default = 'on_default'
grey = 'on_grey'
red = 'on_red'
green = 'on_green'
yellow = 'on_yellow'
blue = 'on_blue'
magenta = 'on_magenta'
cyan = 'on_cyan'
white = 'on_white'
default = "on_default"
grey = "on_grey"
red = "on_red"
green = "on_green"
yellow = "on_yellow"
blue = "on_blue"
magenta = "on_magenta"
cyan = "on_cyan"
white = "on_white"

173
src/cpl_core/console/console.py
View File

@@ -32,7 +32,7 @@ class Console:
_select_menu_items: list[str] = []
_is_first_select_menu_output = True
_selected_menu_item_index: int = 0
_selected_menu_item_char: str = ''
_selected_menu_item_char: str = ""
_selected_menu_option_foreground_color: ForegroundColorEnum = ForegroundColorEnum.default
_selected_menu_option_background_color: BackgroundColorEnum = BackgroundColorEnum.default
_selected_menu_cursor_foreground_color: ForegroundColorEnum = ForegroundColorEnum.default
@@ -60,8 +60,7 @@ class Console:
def set_background_color(cls, color: Union[BackgroundColorEnum, str]):
r"""Sets the background color
Parameter
---------
Parameter:
color: Union[:class:`cpl_core.console.background_color_enum.BackgroundColorEnum`, :class:`str`]
Background color of the console
"""
@@ -74,8 +73,7 @@ class Console:
def set_foreground_color(cls, color: Union[ForegroundColorEnum, str]):
r"""Sets the foreground color
Parameter
---------
Parameter:
color: Union[:class:`cpl_core.console.background_color_enum.BackgroundColorEnum`, :class:`str`]
Foreground color of the console
"""
@@ -94,8 +92,7 @@ class Console:
def set_cursor_position(cls, x: int, y: int):
r"""Sets cursor position
Parameter
---------
Parameter:
x: :class:`int`
X coordinate
y: :class:`int`
@@ -110,8 +107,7 @@ class Console:
def _output(cls, string: str, x: int = None, y: int = None, end: str = None):
r"""Prints given output with given format
Parameter
---------
Parameter:
string: :class:`str`
Message to print
x: :class:`int`
@@ -125,21 +121,30 @@ class Console:
cls._is_first_write = False
if end is None:
end = '\n'
end = "\n"
args = []
colored_args = []
if x is not None and y is not None:
args.append(f'\033[{y};{x}H')
args.append(f"\033[{y};{x}H")
elif cls._x is not None and cls._y is not None:
args.append(f'\033[{cls._y};{cls._x}H')
args.append(f"\033[{cls._y};{cls._x}H")
colored_args.append(string)
if cls._foreground_color != ForegroundColorEnum.default and cls._background_color == BackgroundColorEnum.default:
if (
cls._foreground_color != ForegroundColorEnum.default
and cls._background_color == BackgroundColorEnum.default
):
colored_args.append(cls._foreground_color.value)
elif cls._foreground_color == ForegroundColorEnum.default and cls._background_color != BackgroundColorEnum.default:
elif (
cls._foreground_color == ForegroundColorEnum.default
and cls._background_color != BackgroundColorEnum.default
):
colored_args.append(cls._background_color.value)
elif cls._foreground_color != ForegroundColorEnum.default and cls._background_color != BackgroundColorEnum.default:
elif (
cls._foreground_color != ForegroundColorEnum.default
and cls._background_color != BackgroundColorEnum.default
):
colored_args.append(cls._foreground_color.value)
colored_args.append(cls._background_color.value)
@@ -151,22 +156,23 @@ class Console:
r"""Shows the select menu"""
if not cls._is_first_select_menu_output:
for _ in range(0, len(cls._select_menu_items) + 1):
sys.stdout.write('\x1b[1A\x1b[2K')
sys.stdout.write("\x1b[1A\x1b[2K")
else:
cls._is_first_select_menu_output = False
for i in range(0, len(cls._select_menu_items)):
Console.set_foreground_color(cls._selected_menu_cursor_foreground_color)
Console.set_background_color(cls._selected_menu_cursor_background_color)
placeholder = ''
placeholder = ""
for _ in cls._selected_menu_item_char:
placeholder += ' '
placeholder += " "
Console.write_line(
f'{cls._selected_menu_item_char if cls._selected_menu_item_index == i else placeholder} ')
f"{cls._selected_menu_item_char if cls._selected_menu_item_index == i else placeholder} "
)
Console.set_foreground_color(cls._selected_menu_option_foreground_color)
Console.set_background_color(cls._selected_menu_option_background_color)
Console.write(f'{cls._select_menu_items[i]}')
Console.write(f"{cls._select_menu_items[i]}")
Console.write_line()
@@ -174,12 +180,12 @@ class Console:
def _select_menu_key_press(cls, key):
r"""Event function when key press is detected
Parameter
---------
Parameter:
key: :class:`pynput.keyboard.Key`
Pressed key
"""
from pynput.keyboard import Key
if key == Key.down:
if cls._selected_menu_item_index == len(cls._select_menu_items) - 1:
return
@@ -201,8 +207,7 @@ class Console:
def banner(cls, string: str):
r"""Prints the string as a banner
Parameter
---------
Parameter:
string: :class:`str`
Message to print as banner
"""
@@ -228,7 +233,7 @@ class Console:
cls._hold_back_calls.append(ConsoleCall(cls.clear))
return
os.system('cls' if os.name == 'nt' else 'clear')
os.system("cls" if os.name == "nt" else "clear")
@classmethod
def close(cls):
@@ -241,7 +246,7 @@ class Console:
return
Console.color_reset()
Console.write('\n\n\nPress any key to continue...')
Console.write("\n\n\nPress any key to continue...")
Console.read()
sys.exit()
@@ -254,8 +259,7 @@ class Console:
def error(cls, string: str, tb: str = None):
r"""Prints an error with traceback
Parameter
---------
Parameter:
string: :class:`str`
Error message
tb: :class:`str`
@@ -268,12 +272,12 @@ class Console:
cls._hold_back_calls.append(ConsoleCall(cls.error, string, tb))
return
cls.set_foreground_color('red')
cls.set_foreground_color("red")
if tb is not None:
cls.write_line(f'{string} -> {tb}')
cls.write_line(f"{string} -> {tb}")
else:
cls.write_line(string)
cls.set_foreground_color('default')
cls.set_foreground_color("default")
@classmethod
def enable(cls):
@@ -284,13 +288,11 @@ class Console:
def read(cls, output: str = None) -> str:
r"""Reads in line
Parameter
---------
Parameter:
output: :class:`str`
String to print before input
Returns
-------
Returns:
input()
"""
if output is not None and not cls._hold_back:
@@ -302,22 +304,20 @@ class Console:
def read_line(cls, output: str = None) -> str:
r"""Reads in next line
Parameter
---------
Parameter:
output: :class:`str`
String to print before input
Returns
-------
Returns:
input()
"""
if cls._disabled and not cls._hold_back:
return ''
return ""
if output is not None:
cls.write_line(output)
cls._output('\n', end='')
cls._output("\n", end="")
return input()
@@ -325,8 +325,7 @@ class Console:
def table(cls, header: list[str], values: list[list[str]]):
r"""Prints a table with header and values
Parameter
---------
Parameter:
header: List[:class:`str`]
Header of the table
values: List[List[:class:`str`]]
@@ -342,21 +341,24 @@ class Console:
table = tabulate(values, headers=header)
Console.write_line(table)
Console.write('\n')
Console.write("\n")
@classmethod
def select(cls, char: str, message: str, options: list[str],
header_foreground_color: Union[str, ForegroundColorEnum] = ForegroundColorEnum.default,
header_background_color: Union[str, BackgroundColorEnum] = BackgroundColorEnum.default,
option_foreground_color: Union[str, ForegroundColorEnum] = ForegroundColorEnum.default,
option_background_color: Union[str, BackgroundColorEnum] = BackgroundColorEnum.default,
cursor_foreground_color: Union[str, ForegroundColorEnum] = ForegroundColorEnum.default,
cursor_background_color: Union[str, BackgroundColorEnum] = BackgroundColorEnum.default
) -> str:
def select(
cls,
char: str,
message: str,
options: list[str],
header_foreground_color: Union[str, ForegroundColorEnum] = ForegroundColorEnum.default,
header_background_color: Union[str, BackgroundColorEnum] = BackgroundColorEnum.default,
option_foreground_color: Union[str, ForegroundColorEnum] = ForegroundColorEnum.default,
option_background_color: Union[str, BackgroundColorEnum] = BackgroundColorEnum.default,
cursor_foreground_color: Union[str, ForegroundColorEnum] = ForegroundColorEnum.default,
cursor_background_color: Union[str, BackgroundColorEnum] = BackgroundColorEnum.default,
) -> str:
r"""Prints select menu
Parameter
---------
Parameter:
char: :class:`str`
Character to show which element is selected
message: :class:`str`
@@ -376,8 +378,7 @@ class Console:
cursor_background_color: Union[:class:`str`, :class:`cpl_core.console.background_color_enum.BackgroundColorEnum`]
Background color of the cursor
Returns
-------
Returns:
Selected option as :class:`str`
"""
cls._selected_menu_item_char = char
@@ -396,27 +397,32 @@ class Console:
Console.set_foreground_color(header_foreground_color)
Console.set_background_color(header_background_color)
Console.write_line(message, '\n')
Console.write_line(message, "\n")
cls._show_select_menu()
from pynput import keyboard
with keyboard.Listener(
on_press=cls._select_menu_key_press, suppress=False
) as listener:
with keyboard.Listener(on_press=cls._select_menu_key_press, suppress=False) as listener:
listener.join()
Console.color_reset()
return cls._select_menu_items[cls._selected_menu_item_index]
@classmethod
def spinner(cls, message: str, call: Callable, *args, text_foreground_color: Union[str, ForegroundColorEnum] = None,
spinner_foreground_color: Union[str, ForegroundColorEnum] = None,
text_background_color: Union[str, BackgroundColorEnum] = None,
spinner_background_color: Union[str, BackgroundColorEnum] = None, **kwargs) -> any:
def spinner(
cls,
message: str,
call: Callable,
*args,
text_foreground_color: Union[str, ForegroundColorEnum] = None,
spinner_foreground_color: Union[str, ForegroundColorEnum] = None,
text_background_color: Union[str, BackgroundColorEnum] = None,
spinner_background_color: Union[str, BackgroundColorEnum] = None,
**kwargs,
) -> any:
r"""Shows spinner and calls given function, when function has ended the spinner stops
Parameter
---------
Parameter:
message: :class:`str`
Message of the spinner
call: :class:`Callable`
@@ -434,8 +440,7 @@ class Console:
kwargs: :class:`dict`
Keyword arguments of the call
Returns
-------
Returns:
Return value of call
"""
if cls._hold_back:
@@ -486,11 +491,10 @@ class Console:
return return_value
@classmethod
def write(cls, *args, end=''):
def write(cls, *args, end=""):
r"""Prints in active line
Parameter
---------
Parameter:
args: :class:`list`
Elements to print
end: :class:`str`
@@ -503,15 +507,14 @@ class Console:
cls._hold_back_calls.append(ConsoleCall(cls.write, *args))
return
string = ' '.join(map(str, args))
string = " ".join(map(str, args))
cls._output(string, end=end)
@classmethod
def write_at(cls, x: int, y: int, *args):
r"""Prints at given position
Parameter
---------
Parameter:
x: :class:`int`
X coordinate
y: :class:`int`
@@ -526,15 +529,14 @@ class Console:
cls._hold_back_calls.append(ConsoleCall(cls.write_at, x, y, *args))
return
string = ' '.join(map(str, args))
cls._output(string, x, y, end='')
string = " ".join(map(str, args))
cls._output(string, x, y, end="")
@classmethod
def write_line(cls, *args):
r"""Prints to new line
Parameter
---------
Parameter:
args: :class:`list`
Elements to print
"""
@@ -545,17 +547,16 @@ class Console:
cls._hold_back_calls.append(ConsoleCall(cls.write_line, *args))
return
string = ' '.join(map(str, args))
string = " ".join(map(str, args))
if not cls._is_first_write:
cls._output('')
cls._output(string, end='')
cls._output("")
cls._output(string, end="")
@classmethod
def write_line_at(cls, x: int, y: int, *args):
r"""Prints new line at given position
Parameter
---------
Parameter:
x: :class:`int`
X coordinate
y: :class:`int`
@@ -570,7 +571,7 @@ class Console:
cls._hold_back_calls.append(ConsoleCall(cls.write_line_at, x, y, *args))
return
string = ' '.join(map(str, args))
string = " ".join(map(str, args))
if not cls._is_first_write:
cls._output('', end='')
cls._output(string, x, y, end='')
cls._output("", end="")
cls._output(string, x, y, end="")

3
src/cpl_core/console/console_call.py
View File

@@ -4,8 +4,7 @@ from collections.abc import Callable
class ConsoleCall:
r"""Represents a console call, for hold back when spinner is active
Parameter
---------
Parameter:
function: :class:`Callable`
Function to call
args: :class:`list`

19
src/cpl_core/console/foreground_color_enum.py
View File

@@ -2,13 +2,12 @@ from enum import Enum
class ForegroundColorEnum(Enum):
default = 'default'
grey = 'grey'
red = 'red'
green = 'green'
yellow = 'yellow'
blue = 'blue'
magenta = 'magenta'
cyan = 'cyan'
white = 'white'
default = "default"
grey = "grey"
red = "red"
green = "green"
yellow = "yellow"
blue = "blue"
magenta = "magenta"
cyan = "cyan"
white = "white"

27
src/cpl_core/console/spinner_thread.py
View File

@@ -12,8 +12,7 @@ from cpl_core.console.foreground_color_enum import ForegroundColorEnum
class SpinnerThread(threading.Thread):
r"""Thread to show spinner in terminal
Parameter
---------
Parameter:
msg_len: :class:`int`
Length of the message
foreground_color: :class:`cpl_core.console.foreground_color.ForegroundColorEnum`
@@ -36,7 +35,7 @@ class SpinnerThread(threading.Thread):
def _spinner():
r"""Selects active spinner char"""
while True:
for cursor in '|/-\\':
for cursor in "|/-\\":
yield cursor
def _get_color_args(self) -> list[str]:
@@ -53,37 +52,37 @@ class SpinnerThread(threading.Thread):
def run(self) -> None:
r"""Entry point of thread, shows the spinner"""
columns = 0
if sys.platform == 'win32':
if sys.platform == "win32":
columns = os.get_terminal_size().columns
else:
term_rows, term_columns = os.popen('stty size', 'r').read().split()
term_rows, term_columns = os.popen("stty size", "r").read().split()
columns = int(term_columns)
end_msg = 'done'
end_msg = "done"
end_msg_pos = columns - self._msg_len - len(end_msg)
if end_msg_pos > 0:
print(f'{"" : >{end_msg_pos}}', end='')
print(f'{"" : >{end_msg_pos}}', end="")
else:
print('', end='')
print("", end="")
first = True
spinner = self._spinner()
while self._is_spinning:
if first:
first = False
print(colored(f'{next(spinner): >{len(end_msg) - 1}}', *self._get_color_args()), end='')
print(colored(f"{next(spinner): >{len(end_msg) - 1}}", *self._get_color_args()), end="")
else:
print(colored(f'{next(spinner): >{len(end_msg)}}', *self._get_color_args()), end='')
print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
time.sleep(0.1)
back = ''
back = ""
for i in range(0, len(end_msg)):
back += '\b'
back += "\b"
print(back, end='')
print(back, end="")
sys.stdout.flush()
if not self._exit:
print(colored(end_msg, *self._get_color_args()), end='')
print(colored(end_msg, *self._get_color_args()), end="")
def stop_spinning(self):
r"""Stops the spinner"""

15
src/cpl_core/cpl-core.json
View File

@@ -2,14 +2,14 @@
"ProjectSettings": {
"Name": "cpl-core",
"Version": {
"Major": "2022",
"Minor": "12",
"Micro": "1"
"Major": "2023",
"Minor": "2",
"Micro": "0"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",
"Description": "sh-edraft Common Python library",
"LongDescription": "sh-edraft Common Python library",
"Description": "CPL core",
"LongDescription": "CPL core package",
"URL": "https://www.sh-edraft.de",
"CopyrightDate": "2020 - 2023",
"CopyrightName": "sh-edraft.de",
@@ -31,8 +31,9 @@
"DevDependencies": [
"Sphinx==5.0.2",
"sphinx-rtd-theme==1.0.0",
"myst-parser==0.18.0",
"twine==4.0.2"
"myst-parser==0.18.1",
"twine==4.0.2",
"sphinx-markdown-builder==0.5.5"
],
"PythonVersion": ">=3.10",
"PythonPath": {},

18
src/cpl_core/database/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.database'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.database"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -25,5 +25,5 @@ from .database_settings_name_enum import DatabaseSettingsNameEnum
from .database_settings import DatabaseSettings
from .table_abc import TableABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

18
src/cpl_core/database/connection/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.database.connection'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.database.connection"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -24,5 +24,5 @@ from collections import namedtuple
from .database_connection import DatabaseConnection
from .database_connection_abc import DatabaseConnectionABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

13
src/cpl_core/database/connection/database_connection.py
View File

@@ -1,8 +1,7 @@
from typing import Optional
import mysql.connector as sql
from cpl_core.database.connection.database_connection_abc import \
DatabaseConnectionABC
from cpl_core.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl_core.database.database_settings import DatabaseSettings
from cpl_core.utils.credential_manager import CredentialManager
from mysql.connector.abstracts import MySQLConnectionAbstract
@@ -10,8 +9,7 @@ from mysql.connector.cursor import MySQLCursorBuffered
class DatabaseConnection(DatabaseConnectionABC):
r"""Representation of the database connection
"""
r"""Representation of the database connection"""
def __init__(self):
DatabaseConnectionABC.__init__(self)
@@ -36,10 +34,9 @@ class DatabaseConnection(DatabaseConnectionABC):
charset=database_settings.charset,
use_unicode=database_settings.use_unicode,
buffered=database_settings.buffered,
auth_plugin=database_settings.auth_plugin
auth_plugin=database_settings.auth_plugin,
)
connection.cursor().execute(
f'CREATE DATABASE IF NOT EXISTS `{database_settings.database}`;')
connection.cursor().execute(f"CREATE DATABASE IF NOT EXISTS `{database_settings.database}`;")
self._database = sql.connect(
host=database_settings.host,
port=database_settings.port,
@@ -49,6 +46,6 @@ class DatabaseConnection(DatabaseConnectionABC):
charset=database_settings.charset,
use_unicode=database_settings.use_unicode,
buffered=database_settings.buffered,
auth_plugin=database_settings.auth_plugin
auth_plugin=database_settings.auth_plugin,
)
self._cursor = self._database.cursor()

16
src/cpl_core/database/connection/database_connection_abc.py
View File

@@ -9,22 +9,24 @@ class DatabaseConnectionABC(ABC):
r"""ABC for the :class:`cpl_core.database.connection.database_connection.DatabaseConnection`"""
@abstractmethod
def __init__(self): pass
def __init__(self):
pass
@property
@abstractmethod
def server(self) -> MySQLConnectionAbstract: pass
def server(self) -> MySQLConnectionAbstract:
pass
@property
@abstractmethod
def cursor(self) -> MySQLCursorBuffered: pass
def cursor(self) -> MySQLCursorBuffered:
pass
@abstractmethod
def connect(self, database_settings: DatabaseSettings):
r"""Connects to a database by connection string
Parameter
---------
Parameter:
connection_string: :class:`str`
Database connection string, see: https://docs.sqlalchemy.org/en/14/core/engines.html
"""

18
src/cpl_core/database/context/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.database.context'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.database.context"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -24,5 +24,5 @@ from collections import namedtuple
from .database_context import DatabaseContext
from .database_context_abc import DatabaseContextABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

11
src/cpl_core/database/context/database_context.py
View File

@@ -1,11 +1,9 @@
from typing import Optional
import mysql
from cpl_core.database.connection.database_connection import DatabaseConnection
from cpl_core.database.connection.database_connection_abc import \
DatabaseConnectionABC
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_core.database.table_abc import TableABC
@@ -15,8 +13,7 @@ from mysql.connector.cursor import MySQLCursorBuffered
class DatabaseContext(DatabaseContextABC):
r"""Representation of the database context
Parameter
---------
Parameter:
database_settings: :class:`cpl_core.database.database_settings.DatabaseSettings`
"""
@@ -38,7 +35,7 @@ class DatabaseContext(DatabaseContextABC):
except mysql.connector.Error as err:
# reconnect your cursor as you did in __init__ or wherever
if self._settings is None:
raise Exception('Call DatabaseContext.connect first')
raise Exception("Call DatabaseContext.connect first")
self.connect(self._settings)
def connect(self, database_settings: DatabaseSettings):
@@ -53,7 +50,7 @@ class DatabaseContext(DatabaseContextABC):
def save_changes(self):
self._ping_and_reconnect()
self._db.server.commit()
def select(self, statement: str) -> list[tuple]:
self._ping_and_reconnect()
self._db.cursor.execute(statement)

22
src/cpl_core/database/context/database_context_abc.py
View File

@@ -13,33 +13,31 @@ class DatabaseContextABC(ABC):
@property
@abstractmethod
def cursor(self) -> MySQLCursorBuffered: pass
def cursor(self) -> MySQLCursorBuffered:
pass
@abstractmethod
def connect(self, database_settings: DatabaseSettings):
r"""Connects to a database by connection settings
Parameter
---------
Parameter:
database_settings :class:`cpl_core.database.database_settings.DatabaseSettings`
"""
pass
@abstractmethod
def save_changes(self):
r"""Saves changes of the database"""
pass
@abstractmethod
def select(self, statement: str) -> list[tuple]:
r"""Runs SQL Statements
Parameter
---------
Parameter:
statement: :class:`str`
Returns
-------
Returns:
list: Fetched list of selected elements
"""
pass

15
src/cpl_core/database/database_settings.py
View File

@@ -62,8 +62,7 @@ class DatabaseSettings(ConfigurationModelABC):
def from_dict(self, settings: dict):
r"""Sets attributes from given dict
Parameter
---------
Parameter:
settings: :class:`dict`
"""
try:
@@ -75,20 +74,20 @@ class DatabaseSettings(ConfigurationModelABC):
self._user = settings[DatabaseSettingsNameEnum.user.value]
self._password = settings[DatabaseSettingsNameEnum.password.value]
self._databse = settings[DatabaseSettingsNameEnum.database.value]
if DatabaseSettingsNameEnum.charset.value in settings:
self._charset = settings[DatabaseSettingsNameEnum.charset.value]
if DatabaseSettingsNameEnum.buffered.value in settings:
self._use_unicode = bool(settings[DatabaseSettingsNameEnum.use_unicode.value])
if DatabaseSettingsNameEnum.buffered.value in settings:
self._buffered = bool(settings[DatabaseSettingsNameEnum.buffered.value])
if DatabaseSettingsNameEnum.auth_plugin.value in settings:
self._auth_plugin = settings[DatabaseSettingsNameEnum.auth_plugin.value]
except Exception as e:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.write_line(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')
Console.write_line(f"[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings")
Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
Console.set_foreground_color(ForegroundColorEnum.default)

19
src/cpl_core/database/database_settings_name_enum.py
View File

@@ -2,13 +2,12 @@ from enum import Enum
class DatabaseSettingsNameEnum(Enum):
host = 'Host'
port = 'Port'
user = 'User'
password = 'Password'
database = 'Database'
charset = 'Charset'
use_unicode = 'UseUnicode'
buffered = 'Buffered'
auth_plugin = 'AuthPlugin'
host = "Host"
port = "Port"
user = "User"
password = "Password"
database = "Database"
charset = "Charset"
use_unicode = "UseUnicode"
buffered = "Buffered"
auth_plugin = "AuthPlugin"

20
src/cpl_core/database/table_abc.py
View File

@@ -4,16 +4,15 @@ from typing import Optional
class TableABC(ABC):
@abstractmethod
def __init__(self):
self._created_at: Optional[datetime] = datetime.now().isoformat()
self._modified_at: Optional[datetime] = datetime.now().isoformat()
@property
def created_at(self) -> datetime:
return self._created_at
@property
def modified_at(self) -> datetime:
return self._modified_at
@@ -21,15 +20,18 @@ class TableABC(ABC):
@modified_at.setter
def modified_at(self, value: datetime):
self._modified_at = value
@property
@abstractmethod
def insert_string(self) -> str: pass
def insert_string(self) -> str:
pass
@property
@abstractmethod
def udpate_string(self) -> str: pass
def udpate_string(self) -> str:
pass
@property
@abstractmethod
def delete_string(self) -> str: pass
def delete_string(self) -> str:
pass

18
src/cpl_core/dependency_injection/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.dependency_injection'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.dependency_injection"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -30,5 +30,5 @@ from .service_lifetime_enum import ServiceLifetimeEnum
from .service_provider import ServiceProvider
from .service_provider_abc import ServiceProviderABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

4
src/cpl_core/dependency_injection/scope.py
View File

@@ -1,11 +1,9 @@
from cpl_core.console.console import Console
from cpl_core.dependency_injection.scope_abc import ScopeABC
from cpl_core.dependency_injection.service_provider_abc import ServiceProviderABC
class Scope(ScopeABC):
def __init__(self, service_provider: ServiceProviderABC):
self._service_provider = service_provider
self._service_provider.set_scope(self)
@@ -20,6 +18,6 @@ class Scope(ScopeABC):
@property
def service_provider(self) -> ServiceProviderABC:
return self._service_provider
def dispose(self):
self._service_provider = None

15
src/cpl_core/dependency_injection/scope_abc.py
View File

@@ -1,24 +1,23 @@
from abc import ABC, abstractmethod
class ScopeABC(ABC):
r"""ABC for the class :class:`cpl_core.dependency_injection.scope.Scope`"""
def __init__(self):
pass
@property
@abstractmethod
def service_provider(self):
r"""Returns to service provider of scope
Returns
-------
Returns:
Object of type :class:`cpl_core.dependency_injection.service_provider_abc.ServiceProviderABC`
"""
pass
@abstractmethod
def dispose(self):
r"""Sets service_provider to None
"""
pass
r"""Sets service_provider to None"""
pass

7
src/cpl_core/dependency_injection/scope_builder.py
View File

@@ -5,15 +5,14 @@ from cpl_core.dependency_injection.service_provider_abc import ServiceProviderAB
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
-------
Returns:
Object of type :class:`cpl_core.dependency_injection.scope.Scope`
"""
return Scope(self._service_provider)

2
src/cpl_core/dependency_injection/service_collection.py
View File

@@ -35,7 +35,7 @@ class ServiceCollection(ServiceCollectionABC):
if not isinstance(service, type):
service_type = type(service)
raise Exception(f'Service of type {service_type} already exists')
raise Exception(f"Service of type {service_type} already exists")
self._service_descriptors.append(ServiceDescriptor(service, lifetime, base_type))

40
src/cpl_core/dependency_injection/service_collection_abc.py
View File

@@ -19,8 +19,7 @@ class ServiceCollectionABC(ABC):
def add_db_context(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
r"""Adds database context
Parameter
---------
Parameter:
db_context: Type[:class:`cpl_core.database.context.database_context_abc.DatabaseContextABC`]
Database context
"""
@@ -38,62 +37,56 @@ class ServiceCollectionABC(ABC):
def add_discord(self):
r"""Adds the CPL discord"""
raise NotImplementedError('You should install and use the cpl-discord package')
raise NotImplementedError("You should install and use the cpl-discord package")
pass
def add_translation(self):
r"""Adds the CPL translation"""
raise NotImplementedError('You should install and use the cpl-translation package')
raise NotImplementedError("You should install and use the cpl-translation package")
pass
@abstractmethod
def add_transient(self, service_type: T, service: T = None) -> 'ServiceCollectionABC':
def add_transient(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with transient lifetime
Parameter
---------
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns
------
self: :class:`cpl_core.dependency_injection.service_collection_abc.ServiceCollectionABC
Returns:
self: :class:`cpl_core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
pass
@abstractmethod
def add_scoped(self, service_type: T, service: T = None) -> 'ServiceCollectionABC':
def add_scoped(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with scoped lifetime
Parameter
---------
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns
------
self: :class:`cpl_core.dependency_injection.service_collection_abc.ServiceCollectionABC
Returns:
self: :class:`cpl_core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
pass
@abstractmethod
def add_singleton(self, service_type: T, service: T = None) -> 'ServiceCollectionABC':
def add_singleton(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
r"""Adds a service with singleton lifetime
Parameter
---------
Parameter:
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
Returns
------
self: :class:`cpl_core.dependency_injection.service_collection_abc.ServiceCollectionABC
Returns:
self: :class:`cpl_core.dependency_injection.service_collection_abc.ServiceCollectionABC`
"""
pass
@@ -101,8 +94,7 @@ class ServiceCollectionABC(ABC):
def build_service_provider(self) -> ServiceProviderABC:
r"""Creates instance of the service provider
Returns
-------
Returns:
Object of type :class:`cpl_core.dependency_injection.service_provider_abc.ServiceProviderABC`
"""
pass

4
src/cpl_core/dependency_injection/service_descriptor.py
View File

@@ -7,8 +7,7 @@ from cpl_core.dependency_injection.service_lifetime_enum import ServiceLifetimeE
class ServiceDescriptor:
r"""Descriptor of a service
Parameter
---------
Parameter:
implementation: Union[:class:`type`, Optional[:class:`object`]]
Object or type of service
lifetime: :class:`cpl_core.dependency_injection.service_lifetime_enum.ServiceLifetimeEnum`
@@ -16,7 +15,6 @@ class ServiceDescriptor:
"""
def __init__(self, implementation: Union[type, Optional[object]], lifetime: ServiceLifetimeEnum, base_type=None):
self._service_type = implementation
self._implementation = implementation
self._lifetime = lifetime

1
src/cpl_core/dependency_injection/service_lifetime_enum.py
View File

@@ -2,7 +2,6 @@ from enum import Enum
class ServiceLifetimeEnum(Enum):
singleton = 0
scoped = 1
transient = 2

29
src/cpl_core/dependency_injection/service_provider.py
View File

@@ -18,8 +18,7 @@ from cpl_core.type import T
class ServiceProvider(ServiceProviderABC):
r"""Provider for the services
Parameter
---------
Parameter:
service_descriptors: list[:class:`cpl_core.dependency_injection.service_descriptor.ServiceDescriptor`]
Descriptor of the service
config: :class:`cpl_core.configuration.configuration_abc.ConfigurationABC`
@@ -28,7 +27,12 @@ class ServiceProvider(ServiceProviderABC):
Database representation
"""
def __init__(self, service_descriptors: list[ServiceDescriptor], config: ConfigurationABC, db_context: Optional[DatabaseContextABC]):
def __init__(
self,
service_descriptors: list[ServiceDescriptor],
config: ConfigurationABC,
db_context: Optional[DatabaseContextABC],
):
ServiceProviderABC.__init__(self)
self._service_descriptors: list[ServiceDescriptor] = service_descriptors
@@ -45,7 +49,9 @@ class ServiceProvider(ServiceProviderABC):
def _get_service(self, parameter: Parameter) -> Optional[object]:
for descriptor in self._service_descriptors:
if descriptor.service_type == parameter.annotation or issubclass(descriptor.service_type, parameter.annotation):
if descriptor.service_type == parameter.annotation or issubclass(
descriptor.service_type, parameter.annotation
):
if descriptor.implementation is not None:
return descriptor.implementation
@@ -77,8 +83,7 @@ class ServiceProvider(ServiceProviderABC):
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if parameter.name != "self" and parameter.annotation != Parameter.empty:
if typing.get_origin(parameter.annotation) == list:
params.append(self._get_services(typing.get_args(parameter.annotation)[0]))
@@ -121,7 +126,9 @@ class ServiceProvider(ServiceProviderABC):
self._scope = scope
def create_scope(self) -> ScopeABC:
sb = ScopeBuilder(ServiceProvider(copy.deepcopy(self._service_descriptors), self._configuration, self._database_context))
sb = ScopeBuilder(
ServiceProvider(copy.deepcopy(self._service_descriptors), self._configuration, self._database_context)
)
return sb.build()
def get_service(self, service_type: T) -> Optional[T]:
@@ -134,7 +141,11 @@ class ServiceProvider(ServiceProviderABC):
return result.implementation
implementation = self.build_service(service_type)
if result.lifetime == ServiceLifetimeEnum.singleton or result.lifetime == ServiceLifetimeEnum.scoped and self._scope is not None:
if (
result.lifetime == ServiceLifetimeEnum.singleton
or result.lifetime == ServiceLifetimeEnum.scoped
and self._scope is not None
):
result.implementation = implementation
return implementation
@@ -143,7 +154,7 @@ class ServiceProvider(ServiceProviderABC):
implementations = []
if typing.get_origin(service_type) != list:
raise Exception(f'Invalid type {service_type}! Expected list of type')
raise Exception(f"Invalid type {service_type}! Expected list of type")
implementations.extend(self._get_services(typing.get_args(service_type)[0]))

44
src/cpl_core/dependency_injection/service_provider_abc.py
View File

@@ -10,29 +10,29 @@ from cpl_core.type import T
class ServiceProviderABC(ABC):
r"""ABC for the class :class:`cpl_core.dependency_injection.service_provider.ServiceProvider`"""
_provider: Optional['ServiceProviderABC'] = None
_provider: Optional["ServiceProviderABC"] = None
@abstractmethod
def __init__(self): pass
def __init__(self):
pass
@classmethod
def set_global_provider(cls, provider: 'ServiceProviderABC'):
def set_global_provider(cls, provider: "ServiceProviderABC"):
cls._provider = provider
@abstractmethod
def build_by_signature(self, sig: Signature) -> list[T]: pass
def build_by_signature(self, sig: Signature) -> list[T]:
pass
@abstractmethod
def build_service(self, service_type: type) -> object:
r"""Creates instance of given type
Parameter
---------
Parameter:
instance_type: :class:`type`
The type of the searched instance
Returns
-------
Returns:
Object of the given type
"""
pass
@@ -41,8 +41,7 @@ class ServiceProviderABC(ABC):
def set_scope(self, scope: ScopeABC):
r"""Sets the scope of service provider
Parameter
---------
Parameter:
Object of type :class:`cpl_core.dependency_injection.scope_abc.ScopeABC`
Service scope
"""
@@ -52,23 +51,20 @@ class ServiceProviderABC(ABC):
def create_scope(self) -> ScopeABC:
r"""Creates a service scope
Returns
-------
Returns:
Object of type :class:`cpl_core.dependency_injection.scope_abc.ScopeABC`
"""
pass
@abstractmethod
def get_service(self, instance_type: T) -> Optional[T]:
def get_service(self, instance_type: Type[T]) -> Optional[T]:
r"""Returns instance of given type
Parameter
---------
Parameter:
instance_type: :class:`cpl_core.type.T`
The type of the searched instance
Returns
-------
Returns:
Object of type Optional[:class:`cpl_core.type.T`]
"""
pass
@@ -77,13 +73,11 @@ class ServiceProviderABC(ABC):
def get_services(self, service_type: T) -> list[Optional[T]]:
r"""Returns instance of given type
Parameter
---------
Parameter:
instance_type: :class:`cpl_core.type.T`
The type of the searched instance
Returns
-------
Returns:
Object of type list[Optional[:class:`cpl_core.type.T`]
"""
pass
@@ -92,12 +86,10 @@ class ServiceProviderABC(ABC):
def inject(cls, f=None):
r"""Decorator to allow injection into static and class methods
Parameter
---------
Parameter:
f: Callable
Returns
-------
Returns:
function
"""
if f is None:
@@ -106,7 +98,7 @@ class ServiceProviderABC(ABC):
@functools.wraps(f)
def inner(*args, **kwargs):
if cls._provider is None:
raise Exception(f'{cls.__name__} not build!')
raise Exception(f"{cls.__name__} not build!")
injection = cls._provider.build_by_signature(signature(f))
return f(*injection, *args, **kwargs)

18
src/cpl_core/environment/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.environment'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.environment"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -25,5 +25,5 @@ from .application_environment_abc import ApplicationEnvironmentABC
from .environment_name_enum import EnvironmentNameEnum
from .application_environment import ApplicationEnvironment
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

13
src/cpl_core/environment/application_environment.py
View File

@@ -10,8 +10,7 @@ from cpl_core.environment.environment_name_enum import EnvironmentNameEnum
class ApplicationEnvironment(ApplicationEnvironmentABC):
r"""Represents environment of the application
Parameter
---------
Parameter:
name: :class:`cpl_core.environment.environment_name_enum.EnvironmentNameEnum`
"""
@@ -37,7 +36,7 @@ class ApplicationEnvironment(ApplicationEnvironmentABC):
@property
def application_name(self) -> str:
return self._app_name if self._app_name is not None else ''
return self._app_name if self._app_name is not None else ""
@application_name.setter
def application_name(self, application_name: str):
@@ -45,7 +44,7 @@ class ApplicationEnvironment(ApplicationEnvironmentABC):
@property
def customer(self) -> str:
return self._customer if self._customer is not None else ''
return self._customer if self._customer is not None else ""
@customer.setter
def customer(self, customer: str):
@@ -80,17 +79,17 @@ class ApplicationEnvironment(ApplicationEnvironmentABC):
return str(self._runtime_directory)
def set_runtime_directory(self, runtime_directory: str):
if runtime_directory != '':
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 != '':
if working_directory != "":
self._working_directory = working_directory
os.chdir(self._working_directory)
return
self._working_directory = os.path.abspath('./')
self._working_directory = os.path.abspath("./")
os.chdir(self._working_directory)

48
src/cpl_core/environment/application_environment_abc.py
View File

@@ -11,66 +11,79 @@ class ApplicationEnvironmentABC(ABC):
@property
@abstractmethod
def environment_name(self) -> str: pass
def environment_name(self) -> str:
pass
@environment_name.setter
@abstractmethod
def environment_name(self, environment_name: str): pass
def environment_name(self, environment_name: str):
pass
@property
@abstractmethod
def application_name(self) -> str: pass
def application_name(self) -> str:
pass
@application_name.setter
@abstractmethod
def application_name(self, application_name: str): pass
def application_name(self, application_name: str):
pass
@property
@abstractmethod
def customer(self) -> str: pass
def customer(self) -> str:
pass
@customer.setter
@abstractmethod
def customer(self, customer: str): pass
def customer(self, customer: str):
pass
@property
@abstractmethod
def host_name(self) -> str: pass
def host_name(self) -> str:
pass
@property
@abstractmethod
def start_time(self) -> datetime: pass
def start_time(self) -> datetime:
pass
@start_time.setter
@abstractmethod
def start_time(self, start_time: datetime): pass
def start_time(self, start_time: datetime):
pass
@property
@abstractmethod
def end_time(self): pass
def end_time(self):
pass
@end_time.setter
@abstractmethod
def end_time(self, end_time: datetime): pass
def end_time(self, end_time: datetime):
pass
@property
@abstractmethod
def date_time_now(self) -> datetime: pass
def date_time_now(self) -> datetime:
pass
@property
@abstractmethod
def working_directory(self) -> str: pass
def working_directory(self) -> str:
pass
@property
@abstractmethod
def runtime_directory(self) -> str: pass
def runtime_directory(self) -> str:
pass
@abstractmethod
def set_runtime_directory(self, runtime_directory: str):
r"""Sets the current runtime directory
Parameter
---------
Parameter:
runtime_directory: :class:`str`
Path of the runtime directory
"""
@@ -80,8 +93,7 @@ class ApplicationEnvironmentABC(ABC):
def set_working_directory(self, working_directory: str):
r"""Sets the current working directory
Parameter
---------
Parameter:
working_directory: :class:`str`
Path of the current working directory
"""

9
src/cpl_core/environment/environment_name_enum.py
View File

@@ -2,8 +2,7 @@ from enum import Enum
class EnvironmentNameEnum(Enum):
production = 'production'
staging = 'staging'
testing = 'testing'
development = 'development'
production = "production"
staging = "staging"
testing = "testing"
development = "development"

18
src/cpl_core/logging/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.logging'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.logging"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -27,5 +27,5 @@ from .logging_level_enum import LoggingLevelEnum
from .logging_settings import LoggingSettings
from .logging_settings_name_enum import LoggingSettingsNameEnum
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

21
src/cpl_core/logging/logger_abc.py
View File

@@ -12,8 +12,7 @@ class LoggerABC(ABC):
def header(self, string: str):
r"""Writes a header message
Parameter
---------
Parameter:
string: :class:`str`
String to write as header
"""
@@ -23,8 +22,7 @@ class LoggerABC(ABC):
def trace(self, name: str, message: str):
r"""Writes a trace message
Parameter
---------
Parameter:
name: :class:`str`
Message name
message: :class:`str`
@@ -36,8 +34,7 @@ class LoggerABC(ABC):
def debug(self, name: str, message: str):
r"""Writes a debug message
Parameter
---------
Parameter:
name: :class:`str`
Message name
message: :class:`str`
@@ -49,8 +46,7 @@ class LoggerABC(ABC):
def info(self, name: str, message: str):
r"""Writes an information
Parameter
---------
Parameter:
name: :class:`str`
Message name
message: :class:`str`
@@ -62,8 +58,7 @@ class LoggerABC(ABC):
def warn(self, name: str, message: str):
r"""Writes an warning
Parameter
---------
Parameter:
name: :class:`str`
Message name
message: :class:`str`
@@ -75,8 +70,7 @@ class LoggerABC(ABC):
def error(self, name: str, message: str, ex: Exception = None):
r"""Writes an error
Parameter
---------
Parameter:
name: :class:`str`
Error name
message: :class:`str`
@@ -90,8 +84,7 @@ class LoggerABC(ABC):
def fatal(self, name: str, message: str, ex: Exception = None):
r"""Writes an error and ends the program
Parameter
---------
Parameter:
name: :class:`str`
Error name
message: :class:`str`

70
src/cpl_core/logging/logger_service.py
View File

@@ -17,8 +17,7 @@ from cpl_core.time.time_format_settings import TimeFormatSettings
class Logger(LoggerABC):
r"""Service for logging
Parameter
---------
Parameter:
logging_settings: :class:`cpl_core.logging.logging_settings.LoggingSettings`
Settings for the logger
time_format: :class:`cpl_core.time.time_format_settings.TimeFormatSettings`
@@ -27,7 +26,9 @@ class Logger(LoggerABC):
Environment of the application
"""
def __init__(self, logging_settings: LoggingSettings, time_format: TimeFormatSettings, env: ApplicationEnvironmentABC):
def __init__(
self, logging_settings: LoggingSettings, time_format: TimeFormatSettings, env: ApplicationEnvironmentABC
):
LoggerABC.__init__(self)
self._env = env
@@ -48,7 +49,7 @@ class Logger(LoggerABC):
date_time_now=self._env.date_time_now.strftime(self._time_format_settings.date_time_format),
date_now=self._env.date_time_now.strftime(self._time_format_settings.date_format),
time_now=self._env.date_time_now.strftime(self._time_format_settings.time_format),
start_time=self._env.start_time.strftime(self._time_format_settings.date_time_log_format)
start_time=self._env.start_time.strftime(self._time_format_settings.date_time_log_format),
)
@property
@@ -57,38 +58,36 @@ class Logger(LoggerABC):
date_time_now=self._env.date_time_now.strftime(self._time_format_settings.date_time_format),
date_now=self._env.date_time_now.strftime(self._time_format_settings.date_format),
time_now=self._env.date_time_now.strftime(self._time_format_settings.time_format),
start_time=self._env.start_time.strftime(self._time_format_settings.date_time_log_format)
start_time=self._env.start_time.strftime(self._time_format_settings.date_time_log_format),
)
def _check_for_settings(self, settings: ConfigurationModelABC, settings_type: type):
self._level = LoggingLevelEnum.OFF
self._console = LoggingLevelEnum.FATAL
if settings is None:
self.fatal(__name__, f'Configuration for {settings_type} not found')
self.fatal(__name__, f"Configuration for {settings_type} not found")
def _get_datetime_now(self) -> str:
r"""Returns the date and time by given format
Returns
-------
Returns:
Date and time in given format
"""
try:
return datetime.datetime.now().strftime(self._time_format_settings.date_time_format)
except Exception as e:
self.error(__name__, 'Cannot get time', ex=e)
self.error(__name__, "Cannot get time", ex=e)
def _get_date(self) -> str:
r"""Returns the date by given format
Returns
-------
Returns:
Date in given format
"""
try:
return datetime.datetime.now().strftime(self._time_format_settings.date_format)
except Exception as e:
self.error(__name__, 'Cannot get date', ex=e)
self.error(__name__, "Cannot get date", ex=e)
def create(self) -> None:
r"""Creates path tree and logfile"""
@@ -99,46 +98,44 @@ class Logger(LoggerABC):
if not os.path.exists(self._path):
os.makedirs(self._path)
except Exception as e:
self._fatal_console(__name__, 'Cannot create log dir', ex=e)
self._fatal_console(__name__, "Cannot create log dir", ex=e)
""" create new log file """
try:
# open log file, create if not exists
path = f'{self._path}{self._log}'
permission = 'a+'
path = f"{self._path}{self._log}"
permission = "a+"
if not os.path.isfile(path):
permission = 'w+'
permission = "w+"
f = open(path, permission)
Console.write_line(f'[{__name__}]: Using log file: {path}')
Console.write_line(f"[{__name__}]: Using log file: {path}")
f.close()
except Exception as e:
self._fatal_console(__name__, 'Cannot open log file', ex=e)
self._fatal_console(__name__, "Cannot open log file", ex=e)
def _append_log(self, string: str):
r"""Writes to logfile
Parameter
---------
Parameter:
string: :class:`str`
"""
try:
# open log file and append always
if not os.path.isdir(self._path):
self._warn_console(__name__, 'Log directory not found, try to recreate logger')
self._warn_console(__name__, "Log directory not found, try to recreate logger")
self.create()
with open(self._path + self._log, "a+", encoding="utf-8") as f:
f.write(string + '\n')
f.write(string + "\n")
f.close()
except Exception as e:
self._fatal_console(__name__, f'Cannot append log file, message: {string}', ex=e)
self._fatal_console(__name__, f"Cannot append log file, message: {string}", ex=e)
def _get_string(self, name: str, level: LoggingLevelEnum, message: str) -> str:
r"""Returns input as log entry format
Parameter
---------
Parameter:
name: :class:`str`
Name of the message
level: :class:`cpl_core.logging.logging_level_enum.LoggingLevelEnum`
@@ -146,12 +143,11 @@ class Logger(LoggerABC):
message: :class:`str`
Log message
Returns
-------
Returns:
Formatted string for logging
"""
log_level = level.name
return f'<{self._get_datetime_now()}> [ {log_level} ] [ {name} ]: {message}'
return f"<{self._get_datetime_now()}> [ {log_level} ] [ {name} ]: {message}"
def header(self, string: str):
# append log and print message
@@ -213,11 +209,11 @@ class Logger(LoggerABC):
Console.set_foreground_color(ForegroundColorEnum.default)
def error(self, name: str, message: str, ex: Exception = None):
output = ''
output = ""
if ex is not None:
tb = traceback.format_exc()
self.error(name, message)
output = self._get_string(name, LoggingLevelEnum.ERROR, f'{ex} -> {tb}')
output = self._get_string(name, LoggingLevelEnum.ERROR, f"{ex} -> {tb}")
else:
output = self._get_string(name, LoggingLevelEnum.ERROR, message)
@@ -232,11 +228,11 @@ class Logger(LoggerABC):
Console.set_foreground_color(ForegroundColorEnum.default)
def fatal(self, name: str, message: str, ex: Exception = None):
output = ''
output = ""
if ex is not None:
tb = traceback.format_exc()
self.error(name, message)
output = self._get_string(name, LoggingLevelEnum.FATAL, f'{ex} -> {tb}')
output = self._get_string(name, LoggingLevelEnum.FATAL, f"{ex} -> {tb}")
else:
output = self._get_string(name, LoggingLevelEnum.FATAL, message)
@@ -255,8 +251,7 @@ class Logger(LoggerABC):
def _warn_console(self, name: str, message: str):
r"""Writes a warning to console only
Parameter
---------
Parameter:
name: :class:`str`
Error name
message: :class:`str`
@@ -271,8 +266,7 @@ class Logger(LoggerABC):
def _fatal_console(self, name: str, message: str, ex: Exception = None):
r"""Writes an error to console only
Parameter
---------
Parameter:
name: :class:`str`
Error name
message: :class:`str`
@@ -280,11 +274,11 @@ class Logger(LoggerABC):
ex: :class:`Exception`
Thrown exception
"""
output = ''
output = ""
if ex is not None:
tb = traceback.format_exc()
self.error(name, message)
output = self._get_string(name, LoggingLevelEnum.ERROR, f'{ex} -> {tb}')
output = self._get_string(name, LoggingLevelEnum.ERROR, f"{ex} -> {tb}")
else:
output = self._get_string(name, LoggingLevelEnum.ERROR, message)

15
src/cpl_core/logging/logging_level_enum.py
View File

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

4
src/cpl_core/logging/logging_settings.py
View File

@@ -58,6 +58,6 @@ class LoggingSettings(ConfigurationModelABC):
self._level = LoggingLevelEnum[settings[LoggingSettingsNameEnum.file_level.value]]
except Exception as e:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.write_line(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')
Console.write_line(f"[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings")
Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
Console.set_foreground_color(ForegroundColorEnum.default)

9
src/cpl_core/logging/logging_settings_name_enum.py
View File

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

18
src/cpl_core/mailing/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.mailing'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.mailing"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -27,5 +27,5 @@ from .email_client_abc import EMailClientABC
from .email_client_settings import EMailClientSettings
from .email_client_settings_name_enum import EMailClientSettingsNameEnum
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

45
src/cpl_core/mailing/email.py
View File

@@ -4,8 +4,7 @@ import re
class EMail:
r"""Represents an email
Parameter
---------
Parameter:
header: list[:class:`str`]
Header of the E-Mail
subject: :class:`str`
@@ -18,8 +17,14 @@ class EMail:
Receiver of the E-Mail
"""
def __init__(self, header: list[str] = None, subject: str = None, body: str = None, transceiver: str = None,
receiver: list[str] = None):
def __init__(
self,
header: list[str] = None,
subject: str = None,
body: str = None,
transceiver: str = None,
receiver: list[str] = None,
):
self._header: list[str] = header
self._subject: str = subject
@@ -30,7 +35,7 @@ class EMail:
@property
def header(self) -> str:
return '\r\n'.join(self._header)
return "\r\n".join(self._header)
@property
def header_list(self) -> list[str]:
@@ -65,11 +70,11 @@ class EMail:
if self.check_mail(transceiver):
self._transceiver = transceiver
else:
raise Exception(f'Invalid email: {transceiver}')
raise Exception(f"Invalid email: {transceiver}")
@property
def receiver(self) -> str:
return ','.join(self._receiver)
return ",".join(self._receiver)
@property
def receiver_list(self) -> list[str]:
@@ -83,22 +88,19 @@ class EMail:
def check_mail(address: str) -> bool:
r"""Checks if an email is valid
Parameter
---------
Parameter:
address: :class:`str`
The address to check
Returns
-------
Returns:
Result if E-Mail is valid or not
"""
return bool(re.search('^\\w+([.-]?\\w+)*@\\w+([.-]?\\w+)*(.\\w{2,3})+$', address))
return bool(re.search("^\\w+([.-]?\\w+)*@\\w+([.-]?\\w+)*(.\\w{2,3})+$", address))
def add_header(self, header: str):
r"""Adds header
Parameter
---------
Parameter:
header: :class:`str`
The header of the E-Mail
"""
@@ -110,8 +112,7 @@ class EMail:
def add_receiver(self, receiver: str):
r"""Adds receiver
Parameter
---------
Parameter:
receiver: :class:`str`
The receiver of the E-Mail
"""
@@ -121,20 +122,18 @@ class EMail:
if self.check_mail(receiver):
self._receiver.append(receiver)
else:
raise Exception(f'Invalid email: {receiver}')
raise Exception(f"Invalid email: {receiver}")
def get_content(self, transceiver: str):
r"""Returns the mail as string
Parameter
---------
Parameter:
transceiver: :class:`str`
The transceiver of the E-Mail
Returns
-------
Returns:
E-Mail as string
"""
return str(
f'From: {transceiver}\r\nTo: {self.receiver}\r\n{self.header}\r\nSubject: {self.subject}\r\n{self.body}').encode(
'utf-8')
f"From: {transceiver}\r\nTo: {self.receiver}\r\n{self.header}\r\nSubject: {self.subject}\r\n{self.body}"
).encode("utf-8")

3
src/cpl_core/mailing/email_client_abc.py
View File

@@ -19,8 +19,7 @@ class EMailClientABC(ABC):
def send_mail(self, email: EMail):
r"""Sends email
Parameter
---------
Parameter:
email: :class:`cpl_core.mailing.email.EMail`
Object of the E-Mail to send
"""

55
src/cpl_core/mailing/email_client_service.py
View File

@@ -13,8 +13,7 @@ from cpl_core.utils.credential_manager import CredentialManager
class EMailClient(EMailClientABC):
r"""Service to send emails
Parameter
---------
Parameter:
environment: :class:`cpl_core.environment.application_environment_abc.ApplicationEnvironmentABC`
Environment of the application
logger: :class:`cpl_core.logging.logger_abc.LoggerABC`
@@ -36,44 +35,54 @@ class EMailClient(EMailClientABC):
def create(self):
r"""Creates connection"""
self._logger.trace(__name__, f'Started {__name__}.create')
self._logger.trace(__name__, f"Started {__name__}.create")
self.connect()
self._logger.trace(__name__, f'Stopped {__name__}.create')
self._logger.trace(__name__, f"Stopped {__name__}.create")
def connect(self):
self._logger.trace(__name__, f'Started {__name__}.connect')
self._logger.trace(__name__, f"Started {__name__}.connect")
try:
self._logger.debug(__name__, f'Try to connect to {self._mail_settings.host}:{self._mail_settings.port}')
self._logger.debug(__name__, f"Try to connect to {self._mail_settings.host}:{self._mail_settings.port}")
self._server = SMTP(self._mail_settings.host, self._mail_settings.port)
self._logger.info(__name__, f'Connected to {self._mail_settings.host}:{self._mail_settings.port}')
self._logger.info(__name__, f"Connected to {self._mail_settings.host}:{self._mail_settings.port}")
self._logger.debug(__name__, 'Try to start tls')
self._logger.debug(__name__, "Try to start tls")
self._server.starttls(context=ssl.create_default_context())
self._logger.info(__name__, 'Started tls')
self._logger.info(__name__, "Started tls")
except Exception as e:
self._logger.error(__name__, 'Cannot connect to mail server', e)
self._logger.error(__name__, "Cannot connect to mail server", e)
self._logger.trace(__name__, f'Stopped {__name__}.connect')
self._logger.trace(__name__, f"Stopped {__name__}.connect")
def login(self):
r"""Login to server"""
self._logger.trace(__name__, f'Started {__name__}.login')
self._logger.trace(__name__, f"Started {__name__}.login")
try:
self._logger.debug(__name__, f'Try to login {self._mail_settings.user_name}@{self._mail_settings.host}:{self._mail_settings.port}')
self._server.login(self._mail_settings.user_name, CredentialManager.decrypt(self._mail_settings.credentials))
self._logger.info(__name__, f'Logged on as {self._mail_settings.user_name} to {self._mail_settings.host}:{self._mail_settings.port}')
self._logger.debug(
__name__,
f"Try to login {self._mail_settings.user_name}@{self._mail_settings.host}:{self._mail_settings.port}",
)
self._server.login(
self._mail_settings.user_name, CredentialManager.decrypt(self._mail_settings.credentials)
)
self._logger.info(
__name__,
f"Logged on as {self._mail_settings.user_name} to {self._mail_settings.host}:{self._mail_settings.port}",
)
except Exception as e:
self._logger.error(__name__, 'Cannot login to mail server', e)
self._logger.error(__name__, "Cannot login to mail server", e)
self._logger.trace(__name__, f'Stopped {__name__}.login')
self._logger.trace(__name__, f"Stopped {__name__}.login")
def send_mail(self, email: EMail):
self._logger.trace(__name__, f'Started {__name__}.send_mail')
self._logger.trace(__name__, f"Started {__name__}.send_mail")
try:
self.login()
self._logger.debug(__name__, f'Try to send email to {email.receiver_list}')
self._server.sendmail(self._mail_settings.user_name, email.receiver_list, email.get_content(self._mail_settings.user_name))
self._logger.info(__name__, f'Sent email to {email.receiver_list}')
self._logger.debug(__name__, f"Try to send email to {email.receiver_list}")
self._server.sendmail(
self._mail_settings.user_name, email.receiver_list, email.get_content(self._mail_settings.user_name)
)
self._logger.info(__name__, f"Sent email to {email.receiver_list}")
except Exception as e:
self._logger.error(__name__, f'Cannot send mail to {email.receiver_list}', e)
self._logger.trace(__name__, f'Stopped {__name__}.send_mail')
self._logger.error(__name__, f"Cannot send mail to {email.receiver_list}", e)
self._logger.trace(__name__, f"Stopped {__name__}.send_mail")

11
src/cpl_core/mailing/email_client_settings.py
View File

@@ -11,10 +11,10 @@ class EMailClientSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)
self._host: str = ''
self._host: str = ""
self._port: int = 0
self._user_name: str = ''
self._credentials: str = ''
self._user_name: str = ""
self._credentials: str = ""
@property
def host(self) -> str:
@@ -55,6 +55,5 @@ class EMailClientSettings(ConfigurationModelABC):
self._user_name = settings[EMailClientSettingsNameEnum.user_name.value]
self._credentials = settings[EMailClientSettingsNameEnum.credentials.value]
except Exception as e:
Console.error(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.error(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')
Console.error(f"[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings")
Console.error(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")

9
src/cpl_core/mailing/email_client_settings_name_enum.py
View File

@@ -2,8 +2,7 @@ from enum import Enum
class EMailClientSettingsNameEnum(Enum):
host = 'Host'
port = 'Port'
user_name = 'UserName'
credentials = 'Credentials'
host = "Host"
port = "Port"
user_name = "UserName"
credentials = "Credentials"

18
src/cpl_core/pipes/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.pipes'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.pipes"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -29,5 +29,5 @@ from .pipe_abc import PipeABC
from .to_camel_case_pipe import ToCamelCasePipe
from .to_snake_case_pipe import ToSnakeCasePipe
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

6
src/cpl_core/pipes/bool_pipe.py
View File

@@ -2,8 +2,8 @@ from cpl_core.pipes.pipe_abc import PipeABC
class BoolPipe(PipeABC):
def __init__(self): pass
def __init__(self):
pass
def transform(self, value: bool, *args):
return 'True' if value else 'False'
return "True" if value else "False"

12
src/cpl_core/pipes/first_char_to_lower_pipe.py
View File

@@ -2,19 +2,17 @@ from cpl_core.pipes.pipe_abc import PipeABC
class FirstCharToLowerPipe(PipeABC):
def __init__(self): pass
def __init__(self):
pass
def transform(self, value: any, *args):
r"""Converts first char to lower
Parameter
---------
Parameter:
value: :class:`str`
String to convert
Returns
-------
Returns:
String with first char as lower
"""
return f'{value[0].lower()}{value[1:]}'
return f"{value[0].lower()}{value[1:]}"

12
src/cpl_core/pipes/first_to_upper_pipe.py
View File

@@ -2,19 +2,17 @@ from cpl_core.pipes.pipe_abc import PipeABC
class FirstToUpperPipe(PipeABC):
def __init__(self): pass
def __init__(self):
pass
def transform(self, value: str, *args):
r"""Converts first char to upper
Parameter
---------
Parameter:
chars: :class:`str`
String to convert
Returns
-------
Returns:
String with first char as upper
"""
return f'{value[0].upper()}{value[1:]}'
return f"{value[0].upper()}{value[1:]}"

12
src/cpl_core/pipes/ip_address_pipe.py
View File

@@ -2,23 +2,23 @@ from cpl_core.pipes.pipe_abc import PipeABC
class IPAddressPipe(PipeABC):
def __init__(self): pass
def __init__(self):
pass
def transform(self, value: list[int], *args):
string = ""
if len(value) != 4:
raise Exception('Invalid IP')
raise Exception("Invalid IP")
for i in range(0, len(value)):
byte = value[i]
if byte > 255:
raise Exception('Invalid IP')
raise Exception("Invalid IP")
if i == len(value) - 1:
string += f'{byte}'
string += f"{byte}"
else:
string += f'{byte}.'
string += f"{byte}."
return string

9
src/cpl_core/pipes/pipe_abc.py
View File

@@ -2,9 +2,10 @@ from abc import ABC, abstractmethod
class PipeABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def __init__(self): pass
@abstractmethod
def transform(self, value: any, *args): pass
def transform(self, value: any, *args):
pass

14
src/cpl_core/pipes/to_camel_case_pipe.py
View File

@@ -4,25 +4,23 @@ from cpl_core.pipes import PipeABC
class ToCamelCasePipe(PipeABC):
def __init__(self): pass
def __init__(self):
pass
def transform(self, value: str, *args) -> str:
r"""Converts string to camel case
Parameter
---------
Parameter:
chars: :class:`str`
String to convert
Returns
-------
Returns:
String converted to CamelCase
"""
converted_name = value
char_set = string.punctuation + ' '
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))
converted_name = "".join(word.title() for word in converted_name.split(char))
return converted_name

22
src/cpl_core/pipes/to_snake_case_pipe.py
View File

@@ -4,26 +4,24 @@ from cpl_core.pipes import PipeABC
class ToSnakeCasePipe(PipeABC):
def __init__(self): pass
def __init__(self):
pass
def transform(self, value: str, *args) -> str:
r"""Converts string to snake case
Parameter
---------
Parameter:
chars: :class:`str`
String to convert
Returns
-------
Returns:
String converted to snake_case
"""
# convert to train-case to CamelCase
if '-' in value:
value = ''.join(word.title() for word in value.split('-'))
if "-" in value:
value = "".join(word.title() for word in value.split("-"))
pattern1 = re.compile(r'(.)([A-Z][a-z]+)')
pattern2 = re.compile(r'([a-z0-9])([A-Z])')
file_name = re.sub(pattern1, r'\1_\2', value)
return re.sub(pattern2, r'\1_\2', file_name).lower()
pattern1 = re.compile(r"(.)([A-Z][a-z]+)")
pattern2 = re.compile(r"([a-z0-9])([A-Z])")
file_name = re.sub(pattern1, r"\1_\2", value)
return re.sub(pattern2, r"\1_\2", file_name).lower()

8
src/cpl_core/pipes/version_pipe.py
View File

@@ -3,15 +3,15 @@ from cpl_core.pipes.pipe_abc import PipeABC
class VersionPipe(PipeABC):
def __init__(self): pass
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]}'
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]}'
v_str += f".{value[VersionSettingsNameEnum.micro.value]}"
return v_str

18
src/cpl_core/time/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.time'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.time"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -24,5 +24,5 @@ from collections import namedtuple
from .time_format_settings import TimeFormatSettings
from .time_format_settings_names_enum import TimeFormatSettingsNamesEnum
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

4
src/cpl_core/time/time_format_settings.py
View File

@@ -57,6 +57,6 @@ class TimeFormatSettings(ConfigurationModelABC):
self._date_time_log_format = settings[TimeFormatSettingsNamesEnum.date_time_log_format.value]
except Exception as e:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f'[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings')
Console.write_line(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')
Console.write_line(f"[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings")
Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
Console.set_foreground_color(ForegroundColorEnum.default)

9
src/cpl_core/time/time_format_settings_names_enum.py
View File

@@ -2,8 +2,7 @@ from enum import Enum
class TimeFormatSettingsNamesEnum(Enum):
date_format = 'DateFormat'
time_format = 'TimeFormat'
date_time_format = 'DateTimeFormat'
date_time_log_format = 'DateTimeLogFormat'
date_format = "DateFormat"
time_format = "TimeFormat"
date_time_format = "DateTimeFormat"
date_time_log_format = "DateTimeLogFormat"

2
src/cpl_core/type.py
View File

@@ -1,3 +1,3 @@
from typing import TypeVar
T = TypeVar('T')
T = TypeVar("T")

18
src/cpl_core/utils/__init__.py
View File

@@ -1,21 +1,21 @@
# -*- coding: utf-8 -*-
"""
cpl-core sh-edraft Common Python library
cpl-core CPL core
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
CPL core package
:copyright: (c) 2020 - 2023 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.utils'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2023 sh-edraft.de'
__version__ = '2022.12.1'
__title__ = "cpl_core.utils"
__author__ = "Sven Heidemann"
__license__ = "MIT"
__copyright__ = "Copyright (c) 2020 - 2023 sh-edraft.de"
__version__ = "2023.2.0"
from collections import namedtuple
@@ -25,5 +25,5 @@ from .credential_manager import CredentialManager
from .string import String
from .pip import Pip
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='12', micro='1')
VersionInfo = namedtuple("VersionInfo", "major minor micro")
version_info = VersionInfo(major="2023", minor="2", micro="0")

25
src/cpl_core/utils/credential_manager.py
View File

@@ -8,46 +8,39 @@ class CredentialManager:
def encrypt(string: str) -> str:
r"""Encode with base64
Parameter
---------
Parameter:
string: :class:`str`
String to encode
Returns
-------
Returns:
Encoded string
"""
return base64.b64encode(string.encode('utf-8')).decode('utf-8')
return base64.b64encode(string.encode("utf-8")).decode("utf-8")
@staticmethod
def decrypt(string: str) -> str:
r"""Decode with base64
Parameter
---------
Parameter:
string: :class:`str`
String to decode
Returns
-------
Returns:
Decoded string
"""
return base64.b64decode(string).decode('utf-8')
return base64.b64decode(string).decode("utf-8")
@staticmethod
def build_string(string: str, credentials: str):
r"""Builds string with credentials in it
Parameter
---------
Parameter:
string: :class:`str`
String in which the variable is replaced by credentials
credentials: :class:`str`
String to encode
Returns
-------
Returns:
Decoded string
"""
return string.replace('$credentials', CredentialManager.decrypt(credentials))
return string.replace("$credentials", CredentialManager.decrypt(credentials))

40
src/cpl_core/utils/pip.py
View File

@@ -24,8 +24,7 @@ class Pip:
def set_executable(cls, executable: str):
r"""Sets the executable
Parameter
---------
Parameter:
executable: :class:`str`
The python command
"""
@@ -38,11 +37,11 @@ class Pip:
path = os.path.dirname(os.path.dirname(cls._executable))
cls._env = os.environ
if sys.platform == 'win32':
cls._env['PATH'] = f'{path}\\bin' + os.pathsep + os.environ.get('PATH', '')
if sys.platform == "win32":
cls._env["PATH"] = f"{path}\\bin" + os.pathsep + os.environ.get("PATH", "")
else:
cls._env['PATH'] = f'{path}/bin' + os.pathsep + os.environ.get('PATH', '')
cls._env['VIRTUAL_ENV'] = path
cls._env["PATH"] = f"{path}/bin" + os.pathsep + os.environ.get("PATH", "")
cls._env["VIRTUAL_ENV"] = path
@classmethod
def reset_executable(cls):
@@ -55,26 +54,21 @@ class Pip:
def get_package(cls, package: str) -> Optional[str]:
r"""Gets given package py local pip list
Parameter
---------
Parameter:
package: :class:`str`
Returns
-------
Returns:
The package name as string
"""
result = None
with suppress(Exception):
args = [cls._executable, "-m", "pip", "freeze", "--all"]
result = subprocess.check_output(
args,
stderr=subprocess.DEVNULL, env=cls._env
)
result = subprocess.check_output(args, stderr=subprocess.DEVNULL, env=cls._env)
if result is None:
return None
for p in str(result.decode()).split('\n'):
for p in str(result.decode()).split("\n"):
if p.startswith(package):
return p
@@ -84,8 +78,7 @@ class Pip:
def get_outdated(cls) -> bytes:
r"""Gets table of outdated packages
Returns
-------
Returns:
Bytes string of the command result
"""
args = [cls._executable, "-m", "pip", "list", "--outdated"]
@@ -96,8 +89,7 @@ class Pip:
def install(cls, package: str, *args, source: str = None, stdout=None, stderr=None):
r"""Installs given package
Parameter
---------
Parameter:
package: :class:`str`
The name of the package
args: :class:`list`
@@ -117,7 +109,7 @@ class Pip:
pip_args.append(package)
if source is not None:
pip_args.append(f'--extra-index-url')
pip_args.append(f"--extra-index-url")
pip_args.append(source)
subprocess.run(pip_args, stdout=stdout, stderr=stderr, env=cls._env)
@@ -126,8 +118,7 @@ class Pip:
def uninstall(cls, package: str, stdout=None, stderr=None):
r"""Uninstalls given package
Parameter
---------
Parameter:
package: :class:`str`
The name of the package
stdout: :class:`str`
@@ -137,7 +128,4 @@ class Pip:
"""
args = [cls._executable, "-m", "pip", "uninstall", "--yes", package]
subprocess.run(
args,
stdout=stdout, stderr=stderr, env=cls._env
)
subprocess.run(args, stdout=stdout, stderr=stderr, env=cls._env)

49
src/cpl_core/utils/string.py
View File

@@ -10,20 +10,18 @@ class String:
def convert_to_camel_case(chars: str) -> str:
r"""Converts string to camel case
Parameter
---------
Parameter:
chars: :class:`str`
String to convert
Returns
-------
Returns:
String converted to CamelCase
"""
converted_name = chars
char_set = string.punctuation + ' '
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))
converted_name = "".join(word.title() for word in converted_name.split(char))
return converted_name
@@ -31,61 +29,54 @@ class String:
def convert_to_snake_case(chars: str) -> str:
r"""Converts string to snake case
Parameter
---------
Parameter:
chars: :class:`str`
String to convert
Returns
-------
Returns:
String converted to snake_case
"""
# convert to train-case to CamelCase
if '-' in chars:
chars = ''.join(word.title() for word in chars.split('-'))
if "-" in chars:
chars = "".join(word.title() for word in chars.split("-"))
pattern1 = re.compile(r'(.)([A-Z][a-z]+)')
pattern2 = re.compile(r'([a-z0-9])([A-Z])')
file_name = re.sub(pattern1, r'\1_\2', chars)
return re.sub(pattern2, r'\1_\2', file_name).lower()
pattern1 = re.compile(r"(.)([A-Z][a-z]+)")
pattern2 = re.compile(r"([a-z0-9])([A-Z])")
file_name = re.sub(pattern1, r"\1_\2", chars)
return re.sub(pattern2, r"\1_\2", file_name).lower()
@staticmethod
def first_to_upper(chars: str) -> str:
r"""Converts first char to upper
Parameter
---------
Parameter:
chars: :class:`str`
String to convert
Returns
-------
Returns:
String with first char as upper
"""
return f'{chars[0].upper()}{chars[1:]}'
return f"{chars[0].upper()}{chars[1:]}"
@staticmethod
def first_to_lower(chars: str) -> str:
r"""Converts first char to lower
Parameter
---------
Parameter:
chars: :class:`str`
String to convert
Returns
-------
Returns:
String with first char as lower
"""
return f'{chars[0].lower()}{chars[1:]}'
return f"{chars[0].lower()}{chars[1:]}"
@staticmethod
def random_string(chars: str, length: int) -> str:
r"""Creates random string by given chars and length
Returns
-------
Returns:
String of random chars
"""
return ''.join(random.choice(chars) for _ in range(length))
return "".join(random.choice(chars) for _ in range(length))