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

Renamed project

Browse Source
This commit is contained in:
Sven Heidemann
2021-08-05 14:17:38 +02:00
parent 308e5c9b0c
commit 11241d8f99
63 changed files with 59 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/cpl_core/__init__.py Normal file
View File

@@ -0,0 +1,25 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

29
src/cpl_core/application/__init__.py Normal file
View File

@@ -0,0 +1,29 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.application'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .application_abc import ApplicationABC
from .application_builder import ApplicationBuilder
from .application_builder_abc import ApplicationBuilderABC
from .startup_abc import StartupABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

52
src/cpl_core/application/application_abc.py Normal file
View File

@@ -0,0 +1,52 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.console.console import Console
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.environment import ApplicationEnvironmentABC
class ApplicationABC(ABC):
r"""ABC for the Application class
Parameters
----------
config: :class:`cpl.configuration.configuration_abc.ConfigurationABC`
Contains object loaded from appsettings
services: :class:`cpl.dependency_injection.service_provider_abc.ServiceProviderABC`
Contains instances of prepared objects
"""
@abstractmethod
def __init__(self, config: ConfigurationABC, services: ServiceProviderABC):
self._configuration: Optional[ConfigurationABC] = config
self._environment: Optional[ApplicationEnvironmentABC] = self._configuration.environment
self._services: Optional[ServiceProviderABC] = services
def run(self):
r"""Entry point
Called by custom Application.main
"""
try:
self.configure()
self.main()
except KeyboardInterrupt:
Console.close()
@abstractmethod
def configure(self):
r"""Configure the application
Called by :class:`cpl.application.application_abc.ApplicationABC.run`
"""
pass
@abstractmethod
def main(self):
r"""Custom entry point
Called by :class:`cpl.application.application_abc.ApplicationABC.run`
"""
pass

36
src/cpl_core/application/application_builder.py Normal file
View File

@@ -0,0 +1,36 @@
from typing import Type, Optional
from cpl.application.application_abc import ApplicationABC
from cpl.application.application_builder_abc import ApplicationBuilderABC
from cpl.application.startup_abc import StartupABC
from cpl.configuration.configuration import Configuration
from cpl.dependency_injection.service_collection import ServiceCollection
class ApplicationBuilder(ApplicationBuilderABC):
r"""This is class is used to build a object of :class:`cpl.application.application_abc.ApplicationABC`
Parameter
---------
app: Type[:class:`cpl.application.application_abc.ApplicationABC`]
Application to build
"""
def __init__(self, app: Type[ApplicationABC]):
ApplicationBuilderABC.__init__(self)
self._app = app
self._startup: Optional[StartupABC] = None
self._configuration = Configuration()
self._environment = self._configuration.environment
self._services = ServiceCollection(self._configuration)
def use_startup(self, startup: Type[StartupABC]):
self._startup = startup(self._configuration, self._services)
def build(self) -> ApplicationABC:
if self._startup is not None:
self._startup.configure_configuration()
self._startup.configure_services()
return self._app(self._configuration, self._services.build_service_provider())

34
src/cpl_core/application/application_builder_abc.py Normal file
View File

@@ -0,0 +1,34 @@
from abc import ABC, abstractmethod
from typing import Type
from cpl.application.application_abc import ApplicationABC
from cpl.application.startup_abc import StartupABC
class ApplicationBuilderABC(ABC):
r"""ABC for the :class:`cpl.application.application_builder.ApplicationBuilder`"""
@abstractmethod
def __init__(self, *args):
pass
@abstractmethod
def use_startup(self, startup: Type[StartupABC]):
r"""Sets the custom startup class to use
Parameter
---------
startup: Type[:class:`cpl.application.startup_abc.StartupABC`]
Startup class to use
"""
pass
@abstractmethod
def build(self) -> ApplicationABC:
r"""Creates custom application object
Returns
-------
Object of :class:`cpl.application.application_abc.ApplicationABC`
"""
pass

32
src/cpl_core/application/startup_abc.py Normal file
View File

@@ -0,0 +1,32 @@
from abc import ABC, abstractmethod
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
class StartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self, *args):
pass
@abstractmethod
def configure_configuration(self) -> ConfigurationABC:
r"""Creates configuration of application
Returns
-------
Object of :class:`cpl.configuration.configuration_abc.ConfigurationABC`
"""
pass
@abstractmethod
def configure_services(self) -> ServiceProviderABC:
r"""Creates service provider
Returns
-------
Object of :class:`cpl.dependency_injection.service_provider_abc.ServiceProviderABC`
"""
pass

30
src/cpl_core/configuration/__init__.py Normal file
View File

@@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.configuration'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .configuration import Configuration
from .configuration_abc import ConfigurationABC
from .configuration_model_abc import ConfigurationModelABC
from .configuration_variable_name_enum import ConfigurationVariableNameEnum
from .console_argument import ConsoleArgument
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

397
src/cpl_core/configuration/configuration.py Normal file
View File

@@ -0,0 +1,397 @@
import json
import os
import sys
from collections import Callable
from typing import Union, Type, Optional
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.configuration.configuration_variable_name_enum import ConfigurationVariableNameEnum
from cpl.configuration.console_argument import ConsoleArgument
from cpl.console.console import Console
from cpl.console.foreground_color_enum import ForegroundColorEnum
from cpl.environment.application_environment import ApplicationEnvironment
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.environment.environment_name_enum import EnvironmentNameEnum
class Configuration(ConfigurationABC):
def __init__(self):
r"""Representation of configuration"""
ConfigurationABC.__init__(self)
self._application_environment = ApplicationEnvironment()
self._config: dict[Union[type, str], Union[ConfigurationModelABC, str]] = {}
self._argument_types: list[ConsoleArgument] = []
self._additional_arguments: list[str] = []
self._argument_error_function: Optional[Callable] = None
self._handled_args = []
@property
def environment(self) -> ApplicationEnvironmentABC:
return self._application_environment
@property
def additional_arguments(self) -> list[str]:
return self._additional_arguments
@property
def argument_error_function(self) -> Optional[Callable]:
return self._argument_error_function
@argument_error_function.setter
def argument_error_function(self, argument_error_function: Callable):
self._argument_error_function = argument_error_function
@staticmethod
def _print_info(name: str, message: str):
r"""Prints an info message
Parameter
---------
name: :class:`str`
Info name
message: :class:`str`
Info message
"""
Console.set_foreground_color(ForegroundColorEnum.green)
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
---------
name: :class:`str`
Warning name
message: :class:`str`
Warning message
"""
Console.set_foreground_color(ForegroundColorEnum.yellow)
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
---------
name: :class:`str`
Error name
message: :class:`str`
Error message
"""
Console.set_foreground_color(ForegroundColorEnum.red)
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
---------
name: :class:`str`
Name of the variable
value: :class:`any`
Value of the variable
"""
if name == ConfigurationVariableNameEnum.environment.value:
self._application_environment.environment_name = EnvironmentNameEnum(value)
elif name == ConfigurationVariableNameEnum.name.value:
self._application_environment.application_name = value
elif name == ConfigurationVariableNameEnum.customer.value:
self._application_environment.customer = value
else:
self._config[name] = value
def _validate_argument_by_argument_type(self, argument: str, argument_type: ConsoleArgument,
next_arguments: list[str] = None) -> bool:
r"""Validate argument by argument type
Parameter
---------
argument: :class:`str`
Command as string
argument_type: :class:`cpl.configuration.console_argument.ConsoleArgument`
Command type as ConsoleArgument
next_arguments: list[:class:`str`]
Following arguments of argument
Returns
-------
Object of :class:`bool`
Raises
------
Exception: An error occurred getting an argument for a command
"""
argument_name = ''
value = ''
result = False
if argument_type.value_token != '' and argument_type.value_token in argument:
# ?new=value
found = False
for alias in argument_type.aliases:
if alias in argument:
found = True
if argument_type.name not in argument_name and not found:
return False
if argument_type.is_value_token_optional is not None and argument_type.is_value_token_optional:
if argument_type.name not in self._additional_arguments:
self._additional_arguments.append(argument_type.name)
result = True
if argument_type.token != '' and argument.startswith(argument_type.token):
# --new=value
if len(argument.split(argument_type.token)[1].split(argument_type.value_token)) == 0:
raise Exception(f'Expected argument for command: {argument}')
argument_name = argument.split(argument_type.token)[1].split(argument_type.value_token)[0]
else:
# new=value
argument_name = argument.split(argument_type.value_token)[1]
if argument_name == '':
raise Exception(f'Expected argument for command: {argument_type.name}')
result = True
if argument_type.is_value_token_optional is True:
is_valid = False
name_list = argument.split(argument_type.token)
if len(name_list) > 1:
value_list = name_list[1].split(argument_type.value_token)
if len(value_list) > 1:
is_valid = True
value = argument.split(argument_type.token)[1].split(argument_type.value_token)[1]
if not is_valid:
if argument_type.name not in self._additional_arguments:
self._additional_arguments.append(argument_type.name)
result = True
else:
value = argument.split(argument_type.token)[1].split(argument_type.value_token)[1]
if argument_name != argument_type.name and argument_name not in argument_type.aliases:
return False
self._set_variable(argument_type.name, value)
result = True
elif argument_type.value_token == ' ':
# ?new value
found = False
for alias in argument_type.aliases:
if alias == argument or f' {alias} ' == argument:
found = True
if argument_type.name not in argument and not found:
return False
if (next_arguments is None or len(next_arguments) == 0) and \
argument_type.is_value_token_optional is not True:
raise Exception(f'Expected argument for command: {argument_type.name}')
if (next_arguments is None or len(next_arguments) == 0) and argument_type.is_value_token_optional is True:
value = ''
else:
value = next_arguments[0]
next_arguments.remove(value)
self._handled_args.append(value)
if argument_type.token != '' and argument.startswith(argument_type.token):
# --new value
argument_name = argument.split(argument_type.token)[1]
else:
# new value
argument_name = argument
if argument_name != argument_type.name and argument_name not in argument_type.aliases:
return False
if value == '':
if argument_type.name not in self._additional_arguments:
self._additional_arguments.append(argument_type.name)
else:
self._set_variable(argument_type.name, value)
result = True
elif argument_type.name == argument or argument in argument_type.aliases:
# new
self._additional_arguments.append(argument_type.name)
result = True
if result:
self._handled_args.append(argument)
if next_arguments is not None and len(next_arguments) > 0:
next_args = []
if len(next_arguments) > 1:
next_args = next_arguments[1:]
if argument_type.console_arguments is not None and len(argument_type.console_arguments) > 0:
found_child = False
for child_argument_type in argument_type.console_arguments:
found_child = self._validate_argument_by_argument_type(
next_arguments[0],
child_argument_type,
next_args
)
if found_child and child_argument_type.name not in self._additional_arguments:
self._additional_arguments.append(child_argument_type.name)
if found_child:
break
if not found_child:
result = self._validate_argument_by_argument_type(next_arguments[0], argument_type, next_args)
return result
def _load_json_file(self, file: str, output: bool) -> dict:
r"""Reads the json file
Parameter
---------
file: :class:`str`
Name of the file
output: :class:`bool`
Specifies whether an output should take place
Returns
-------
Object of :class:`dict`
"""
try:
# open config file, create if not exists
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}')
return json_cfg
except Exception as e:
self._print_error(__name__, f'Cannot load config file: {file}! -> {e}')
return {}
def add_environment_variables(self, prefix: str):
for variable in ConfigurationVariableNameEnum.to_list():
var_name = f'{prefix}{variable}'
if var_name in [key.upper() for key in os.environ.keys()]:
self._set_variable(variable, os.environ[var_name])
def add_console_argument(self, argument: ConsoleArgument):
self._argument_types.append(argument)
def add_console_arguments(self, error: bool = None):
for arg_name in ConfigurationVariableNameEnum.to_list():
self.add_console_argument(ConsoleArgument('--', str(arg_name).upper(), [str(arg_name).lower()], '='))
arg_list = sys.argv[1:]
for i in range(0, len(arg_list)):
argument = arg_list[i]
next_arguments = []
error_message = ''
if argument in self._handled_args:
break
if i + 1 < len(arg_list):
next_arguments = arg_list[i + 1:]
found = False
for argument_type in self._argument_types:
try:
found = self._validate_argument_by_argument_type(argument, argument_type, next_arguments)
if found:
break
except Exception as e:
error_message = e
if not found and error_message == '' and error is not False:
error_message = f'Invalid argument: {argument}'
if error_message != '':
if self._argument_error_function is not None:
self._argument_error_function(error_message)
else:
self._print_error(__name__, error_message)
exit()
add_args = []
for next_arg in next_arguments:
if next_arg not in self._handled_args and next_arg not in self._additional_arguments:
add_args.append(next_arg)
self._set_variable(f'{argument}AdditionalArguments', add_args)
def add_json_file(self, name: str, optional: bool = None, output: bool = True, path: str = None):
if os.path.isabs(name):
file_path = name
else:
path_root = self._application_environment.working_directory
if path is not None:
path_root = path
if str(path_root).endswith('/') and not name.startswith('/'):
file_path = f'{path_root}{name}'
else:
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}')
exit()
if output:
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:
configuration = sub()
configuration.from_dict(value)
self.add_configuration(sub, configuration)
def add_configuration(self, key_type: Union[str, type], value: ConfigurationModelABC):
self._config[key_type] = value
def get_configuration(self, search_type: Union[str, Type[ConfigurationModelABC]]) -> \
Union[str, Callable[ConfigurationModelABC]]:
if type(search_type) is str:
if search_type == ConfigurationVariableNameEnum.environment.value:
return self._application_environment.environment_name
elif search_type == ConfigurationVariableNameEnum.name.value:
return self._application_environment.application_name
elif search_type == ConfigurationVariableNameEnum.customer.value:
return self._application_environment.customer
if search_type not in self._config:
return None
for config_model in self._config:
if config_model == search_type:
return self._config[config_model]

109
src/cpl_core/configuration/configuration_abc.py Normal file
View File

@@ -0,0 +1,109 @@
from abc import abstractmethod, ABC
from collections import Callable
from typing import Type, Union, Optional
from cpl.configuration.console_argument import ConsoleArgument
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
class ConfigurationABC(ABC):
@abstractmethod
def __init__(self):
r"""ABC for the :class:`cpl.configuration.configuration.Configuration`"""
pass
@property
@abstractmethod
def environment(self) -> ApplicationEnvironmentABC: pass
@property
@abstractmethod
def additional_arguments(self) -> list[str]: pass
@property
@abstractmethod
def argument_error_function(self) -> Optional[Callable]: pass
@argument_error_function.setter
@abstractmethod
def argument_error_function(self, argument_error_function: Callable): pass
@abstractmethod
def add_environment_variables(self, prefix: str):
r"""Reads the environment variables
Parameter
---------
prefix: :class:`str`
Prefix of the variables
"""
pass
@abstractmethod
def add_console_argument(self, argument: ConsoleArgument):
r"""Adds console argument to known console arguments
Parameter
---------
argument: :class:`cpl.configuration.console_argument.ConsoleArgument`
Specifies the console argument
"""
pass
@abstractmethod
def add_console_arguments(self, error: bool = None):
r"""Reads the console arguments
Parameter
---------
error: :class:`bool`
Defines is invalid argument error will be shown or not
"""
pass
@abstractmethod
def add_json_file(self, name: str, optional: bool = None, output: bool = True, path: str = None):
r"""Reads and saves settings from given json file
Parameter
---------
name: :class:`str`
Name of the file
optional: :class:`str`
Specifies whether an error should occur if the file was not found
output: :class:`bool`
Specifies whether an output should take place
path: :class:`str`
Path in which the file should be stored
"""
pass
@abstractmethod
def add_configuration(self, key_type: Union[str, type], value: Union[str, ConfigurationModelABC]):
r"""Add configuration object
Parameter
---------
key_type: Union[:class:`str`, :class:`type`]
Type of the value
value: Union[:class:`str`, :class:`cpl.configuration.configuration_model_abc.ConfigurationModelABC`]
Object of the value
"""
pass
@abstractmethod
def get_configuration(self, search_type: Union[str, Type[ConfigurationModelABC]]) -> Union[str, Callable[ConfigurationModelABC]]:
r"""Returns value from configuration by given type
Parameter
---------
search_type: Union[:class:`str`, Type[:class:`cpl.configuration.configuration_model_abc.ConfigurationModelABC`]]
Type to search for
Returns
-------
Object of Union[:class:`str`, Callable[:class:`cpl.configuration.configuration_model_abc.ConfigurationModelABC`]]
"""
pass

19
src/cpl_core/configuration/configuration_model_abc.py Normal file
View File

@@ -0,0 +1,19 @@
from abc import ABC, abstractmethod
class ConfigurationModelABC(ABC):
@abstractmethod
def __init__(self):
r"""ABC for settings representation"""
pass
@abstractmethod
def from_dict(self, settings: dict):
r"""Converts attributes to dict
Parameter
---------
settings: :class:`dict`
"""
pass

12
src/cpl_core/configuration/configuration_variable_name_enum.py Normal file
View File

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

51
src/cpl_core/configuration/console_argument.py Normal file
View File

@@ -0,0 +1,51 @@
class ConsoleArgument:
def __init__(self,
token: str,
name: str,
aliases: list[str],
value_token: str,
is_value_token_optional: bool = None,
console_arguments: list['ConsoleArgument'] = None
):
r"""Representation of an console argument
Parameter
---------
token: :class:`str`
name: :class:`str`
aliases: list[:class:`str`]
value_token: :class:`str`
is_value_token_optional: :class:`bool`
console_arguments: List[:class:`cpl.configuration.console_argument.ConsoleArgument`]
"""
self._token = token
self._name = name
self._aliases = aliases
self._value_token = value_token
self._is_value_token_optional = is_value_token_optional
self._console_arguments = console_arguments
@property
def token(self) -> str:
return self._token
@property
def name(self) -> str:
return self._name
@property
def aliases(self) -> list[str]:
return self._aliases
@property
def value_token(self) -> str:
return self._value_token
@property
def is_value_token_optional(self) -> bool:
return self._is_value_token_optional
@property
def console_arguments(self) -> list['ConsoleArgument']:
return self._console_arguments

30
src/cpl_core/console/__init__.py Normal file
View File

@@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.console'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .background_color_enum import BackgroundColorEnum
from .console import Console
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='2021', minor='10', micro='6')

14
src/cpl_core/console/background_color_enum.py Normal file
View File

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

572
src/cpl_core/console/console.py Normal file
View File

@@ -0,0 +1,572 @@
import os
import sys
import time
from collections import Callable
from typing import Union, Optional
import colorama
import pyfiglet
from pynput import keyboard
from pynput.keyboard import Key
from tabulate import tabulate
from termcolor import colored
from cpl.console.background_color_enum import BackgroundColorEnum
from cpl.console.console_call import ConsoleCall
from cpl.console.foreground_color_enum import ForegroundColorEnum
from cpl.console.spinner_thread import SpinnerThread
class Console:
r"""Useful functions for handling with input and output"""
colorama.init()
_is_first_write = True
_background_color: BackgroundColorEnum = BackgroundColorEnum.default
_foreground_color: ForegroundColorEnum = ForegroundColorEnum.default
_x: Optional[int] = None
_y: Optional[int] = None
_disabled: bool = False
_hold_back = False
_hold_back_calls: list[ConsoleCall] = []
_select_menu_items: list[str] = []
_is_first_select_menu_output = True
_selected_menu_item_index: int = 0
_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
_selected_menu_cursor_background_color: BackgroundColorEnum = BackgroundColorEnum.default
"""Properties"""
@classmethod
@property
def background_color(cls) -> str:
return str(cls._background_color.value)
@classmethod
@property
def foreground_color(cls) -> str:
return str(cls._foreground_color.value)
"""Settings"""
@classmethod
def set_hold_back(cls, value: bool):
cls._hold_back = value
@classmethod
def set_background_color(cls, color: Union[BackgroundColorEnum, str]):
r"""Sets the background color
Parameter
---------
color: Union[:class:`cpl.console.background_color_enum.BackgroundColorEnum`, :class:`str`]
Background color of the console
"""
if type(color) is str:
cls._background_color = BackgroundColorEnum[color]
else:
cls._background_color = color
@classmethod
def set_foreground_color(cls, color: Union[ForegroundColorEnum, str]):
r"""Sets the foreground color
Parameter
---------
color: Union[:class:`cpl.console.background_color_enum.BackgroundColorEnum`, :class:`str`]
Foreground color of the console
"""
if type(color) is str:
cls._foreground_color = ForegroundColorEnum[color]
else:
cls._foreground_color = color
@classmethod
def reset_cursor_position(cls):
r"""Resets cursor position"""
cls._x = None
cls._y = None
@classmethod
def set_cursor_position(cls, x: int, y: int):
r"""Sets cursor position
Parameter
---------
x: :class:`int`
X coordinate
y: :class:`int`
Y coordinate
"""
cls._x = x
cls._y = y
"""Useful protected functions"""
@classmethod
def _output(cls, string: str, x: int = None, y: int = None, end: str = None):
r"""Prints given output with given format
Parameter
---------
string: :class:`str`
Message to print
x: :class:`int`
X coordinate
y: :class:`int`
Y coordinate
end: :class:`str`
End character of the message (could be \n)
"""
if cls._is_first_write:
cls._is_first_write = False
if end is None:
end = '\n'
args = []
colored_args = []
if x is not None and y is not None:
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')
colored_args.append(string)
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:
colored_args.append(cls._background_color.value)
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)
args.append(colored(*colored_args))
print(*args, end=end)
@classmethod
def _show_select_menu(cls):
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')
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 = ''
for _ in cls._selected_menu_item_char:
placeholder += ' '
Console.write_line(
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_line()
@classmethod
def _select_menu_key_press(cls, key: Key):
r"""Event function when key press is detected
Parameter
---------
key: :class:`pynput.keyboard.Key`
Pressed key
"""
if key == Key.down:
if cls._selected_menu_item_index == len(cls._select_menu_items) - 1:
return
cls._selected_menu_item_index += 1
cls._show_select_menu()
elif key == Key.up:
if cls._selected_menu_item_index == 0:
return
cls._selected_menu_item_index -= 1
cls._show_select_menu()
elif key == Key.enter:
return False
""" Useful public functions"""
@classmethod
def banner(cls, string: str):
r"""Prints the string as a banner
Parameter
---------
string: :class:`str`
Message to print as banner
"""
if cls._disabled:
return
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.banner, string))
return
ascii_banner = pyfiglet.figlet_format(string)
cls.write_line(ascii_banner)
@classmethod
def color_reset(cls):
r"""Resets the color settings"""
cls._background_color = BackgroundColorEnum.default
cls._foreground_color = ForegroundColorEnum.default
@classmethod
def clear(cls):
r"""Clears the console"""
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.clear))
return
os.system('cls' if os.name == 'nt' else 'clear')
@classmethod
def close(cls):
r"""Closes the application"""
if cls._disabled:
return
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.close))
return
Console.color_reset()
Console.write('\n\n\nPress any key to continue...')
Console.read()
exit()
@classmethod
def disable(cls):
r"""Disables console interaction"""
cls._disabled = True
@classmethod
def error(cls, string: str, tb: str = None):
r"""Prints an error with traceback
Parameter
---------
string: :class:`str`
Error message
tb: :class:`str`
Error traceback
"""
if cls._disabled:
return
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.error, string, tb))
return
cls.set_foreground_color('red')
if tb is not None:
cls.write_line(f'{string} -> {tb}')
else:
cls.write_line(string)
cls.set_foreground_color('default')
@classmethod
def enable(cls):
r"""Enables console interaction"""
cls._disabled = False
@classmethod
def read(cls, output: str = None) -> str:
r"""Reads in line
Parameter
---------
output: :class:`str`
String to print before input
Returns
-------
input()
"""
if output is not None and not cls._hold_back:
cls.write_line(output)
return input()
@classmethod
def read_line(cls, output: str = None) -> str:
r"""Reads in next line
Parameter
---------
output: :class:`str`
String to print before input
Returns
-------
input()
"""
if cls._disabled and not cls._hold_back:
return ''
if output is not None:
cls.write_line(output)
cls._output('\n', end='')
return input()
@classmethod
def table(cls, header: list[str], values: list[list[str]]):
r"""Prints a table with header and values
Parameter
---------
header: List[:class:`str`]
Header of the table
values: List[List[:class:`str`]]
Values of the table
"""
if cls._disabled:
return
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.table, header, values))
return
table = tabulate(values, headers=header)
Console.write_line(table)
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:
r"""Prints select menu
Parameter
---------
char: :class:`str`
Character to show which element is selected
message: :class:`str`
Message or header of the selection
options: List[:class:`str`]
Selectable options
header_foreground_color: Union[:class:`str`, :class:`cpl.console.foreground_color_enum.ForegroundColorEnum`]
Foreground color of the header
header_background_color: Union[:class:`str`, :class:`cpl.console.background_color_enum.BackgroundColorEnum`]
Background color of the header
option_foreground_color: Union[:class:`str`, :class:`cpl.console.foreground_color_enum.ForegroundColorEnum`]
Foreground color of the options
option_background_color: Union[:class:`str`, :class:`cpl.console.background_color_enum.BackgroundColorEnum`]
Background color of the options
cursor_foreground_color: Union[:class:`str`, :class:`cpl.console.foreground_color_enum.ForegroundColorEnum`]
Foreground color of the cursor
cursor_background_color: Union[:class:`str`, :class:`cpl.console.background_color_enum.BackgroundColorEnum`]
Background color of the cursor
Returns
-------
Selected option as :class:`str`
"""
cls._selected_menu_item_char = char
cls.options = options
cls._select_menu_items = cls.options
if option_foreground_color is not None:
cls._selected_menu_option_foreground_color = option_foreground_color
if option_background_color is not None:
cls._selected_menu_option_background_color = option_background_color
if cursor_foreground_color is not None:
cls._selected_menu_cursor_foreground_color = cursor_foreground_color
if cursor_background_color is not None:
cls._selected_menu_cursor_background_color = cursor_background_color
Console.set_foreground_color(header_foreground_color)
Console.set_background_color(header_background_color)
Console.write_line(message, '\n')
cls._show_select_menu()
with keyboard.Listener(
on_press=cls._select_menu_key_press, suppress=True) 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:
r"""Shows spinner and calls given function, when function has ended the spinner stops
Parameter
---------
message: :class:`str`
Message of the spinner
call: :class:`Callable`
Function to call
args: :class:`list`
Arguments of the function
text_foreground_color: Union[:class:`str`, :class:`cpl.console.foreground_color_enum.ForegroundColorEnum`]
Foreground color of the text
spinner_foreground_color: Union[:class:`str`, :class:`cpl.console.foreground_color_enum.ForegroundColorEnum`]
Foreground color of the spinner
text_background_color: Union[:class:`str`, :class:`cpl.console.background_color_enum.BackgroundColorEnum`]
Background color of the text
spinner_background_color: Union[:class:`str`, :class:`cpl.console.background_color_enum.BackgroundColorEnum`]
Background color of the spinner
kwargs: :class:`dict`
Keyword arguments of the call
Returns
-------
Return value of call
"""
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.spinner, message, call, *args))
return
if text_foreground_color is not None:
cls.set_foreground_color(text_foreground_color)
if text_background_color is not None:
cls.set_background_color(text_background_color)
if type(spinner_foreground_color) is str:
spinner_foreground_color = ForegroundColorEnum[spinner_foreground_color]
if type(spinner_background_color) is str:
spinner_background_color = BackgroundColorEnum[spinner_background_color]
cls.write_line(message)
cls.set_hold_back(True)
spinner = SpinnerThread(len(message), spinner_foreground_color, spinner_background_color)
spinner.start()
return_value = None
try:
return_value = call(*args, **kwargs)
except KeyboardInterrupt:
spinner.exit()
cls.close()
spinner.stop_spinning()
cls.set_hold_back(False)
cls.set_foreground_color(ForegroundColorEnum.default)
cls.set_background_color(BackgroundColorEnum.default)
for call in cls._hold_back_calls:
call.function(*call.args)
cls._hold_back_calls = []
time.sleep(0.1)
return return_value
@classmethod
def write(cls, *args, end=''):
r"""Prints in active line
Parameter
---------
args: :class:`list`
Elements to print
end: :class:`str`
Last character to print
"""
if cls._disabled:
return
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.write, args))
return
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
---------
x: :class:`int`
X coordinate
y: :class:`int`
Y coordinate
args: :class:`list`
Elements to print
"""
if cls._disabled:
return
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.write_at, x, y, args))
return
string = ' '.join(map(str, args))
cls._output(string, x, y, end='')
@classmethod
def write_line(cls, *args):
r"""Prints to new line
Parameter
---------
args: :class:`list`
Elements to print
"""
if cls._disabled:
return
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.write_line, args))
return
string = ' '.join(map(str, args))
if not cls._is_first_write:
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
---------
x: :class:`int`
X coordinate
y: :class:`int`
Y coordinate
args: :class:`list`
Elements to print
"""
if cls._disabled:
return
if cls._hold_back:
cls._hold_back_calls.append(ConsoleCall(cls.write_line_at, x, y, args))
return
string = ' '.join(map(str, args))
if not cls._is_first_write:
cls._output('', end='')
cls._output(string, x, y, end='')

25
src/cpl_core/console/console_call.py Normal file
View File

@@ -0,0 +1,25 @@
from collections import Callable
class ConsoleCall:
r"""Represents a console call, for hold back when spinner is active
Parameter
---------
function: :class:`Callable`
Function to call
args: :class:`list`
List of arguments
"""
def __init__(self, function: Callable, *args):
self._func = function
self._args = args
@property
def function(self):
return self._func
@property
def args(self):
return self._args

14
src/cpl_core/console/foreground_color_enum.py Normal file
View File

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

97
src/cpl_core/console/spinner_thread.py Normal file
View File

@@ -0,0 +1,97 @@
import os
import sys
import threading
import time
from termcolor import colored
from cpl.console.background_color_enum import BackgroundColorEnum
from cpl.console.foreground_color_enum import ForegroundColorEnum
class SpinnerThread(threading.Thread):
r"""Thread to show spinner in terminal
Parameter
---------
msg_len: :class:`int`
Length of the message
foreground_color: :class:`cpl.console.foreground_color.ForegroundColorEnum`
Foreground color of the spinner
background_color: :class:`cpl.console.background_color.BackgroundColorEnum`
Background color of the spinner
"""
def __init__(self, msg_len: int, foreground_color: ForegroundColorEnum, background_color: BackgroundColorEnum):
threading.Thread.__init__(self)
self._msg_len = msg_len
self._foreground_color = foreground_color
self._background_color = background_color
self._is_spinning = True
self._exit = False
@staticmethod
def _spinner():
r"""Selects active spinner char"""
while True:
for cursor in '|/-\\':
yield cursor
def _get_color_args(self) -> list[str]:
r"""Creates color arguments"""
color_args = []
if self._foreground_color is not None:
color_args.append(str(self._foreground_color.value))
if self._background_color is not None:
color_args.append(str(self._background_color.value))
return color_args
def run(self) -> None:
r"""Entry point of thread, shows the spinner"""
columns = 0
if sys.platform == 'win32':
columns = os.get_terminal_size().columns
else:
term_rows, term_columns = os.popen('stty size', 'r').read().split()
columns = int(term_columns)
end_msg = 'done'
end_msg_pos = columns - self._msg_len - len(end_msg)
if end_msg_pos > 0:
print(f'{"" : >{end_msg_pos}}', end='')
else:
print('', end='')
first = True
spinner = self._spinner()
while self._is_spinning:
if first:
first = False
print(colored(f'{next(spinner): >{len(end_msg) - 1}}', *self._get_color_args()), end='')
else:
print(colored(f'{next(spinner): >{len(end_msg)}}', *self._get_color_args()), end='')
time.sleep(0.1)
back = ''
for i in range(0, len(end_msg)):
back += '\b'
print(back, end='')
sys.stdout.flush()
if not self._exit:
print(colored(end_msg, *self._get_color_args()), end='')
def stop_spinning(self):
r"""Stops the spinner"""
self._is_spinning = False
time.sleep(0.1)
def exit(self):
r"""Stops the spinner"""
self._is_spinning = False
self._exit = True
time.sleep(0.1)

54
src/cpl_core/cpl_core.json Normal file
View File

@@ -0,0 +1,54 @@
{
"ProjectSettings": {
"Name": "sh_cpl-core",
"Version": {
"Major": "2021",
"Minor": "10",
"Micro": "6"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",
"Description": "sh-edraft Common Python library",
"LongDescription": "sh-edraft Common Python library",
"URL": "https://www.sh-edraft.de",
"CopyrightDate": "2020 - 2021",
"CopyrightName": "sh-edraft.de",
"LicenseName": "MIT",
"LicenseDescription": "MIT, see LICENSE for more details.",
"Dependencies": [
"colorama==0.4.4",
"mysql-connector==2.2.9",
"psutil==5.8.0",
"packaging==20.9",
"pyfiglet==0.8.post1",
"pynput==1.7.3",
"SQLAlchemy==1.4.7",
"setuptools==56.0.0",
"tabulate==0.8.9",
"termcolor==1.1.0",
"watchdog==2.0.2",
"wheel==0.36.2"
],
"PythonVersion": ">=3.8",
"PythonPath": {},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "",
"EntryPoint": "",
"IncludePackageData": true,
"Included": [
"*/templates"
],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

28
src/cpl_core/database/__init__.py Normal file
View File

@@ -0,0 +1,28 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.database'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .database_model import DatabaseModel
from .database_settings import DatabaseSettings
from .database_settings_name_enum import DatabaseSettingsNameEnum
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

27
src/cpl_core/database/connection/__init__.py Normal file
View File

@@ -0,0 +1,27 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.database.connection'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .database_connection import DatabaseConnection
from .database_connection_abc import DatabaseConnectionABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

64
src/cpl_core/database/connection/database_connection.py Normal file
View File

@@ -0,0 +1,64 @@
from typing import Optional
from sqlalchemy import engine, create_engine
from sqlalchemy.orm import Session, sessionmaker
from cpl.console.console import Console
from cpl.console.foreground_color_enum import ForegroundColorEnum
from cpl.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl.database.database_settings import DatabaseSettings
class DatabaseConnection(DatabaseConnectionABC):
r"""Representation of the database connection
Parameter
---------
database_settings: :class:`cpl.database.database_settings.DatabaseSettings`
"""
def __init__(self, database_settings: DatabaseSettings):
DatabaseConnectionABC.__init__(self)
self._db_settings = database_settings
self._engine: Optional[engine] = None
self._session: Optional[Session] = None
self._credentials: Optional[str] = None
@property
def engine(self) -> engine:
return self._engine
@property
def session(self) -> Session:
return self._session
def connect(self, connection_string: str):
try:
self._engine = create_engine(connection_string)
if self._db_settings.auth_plugin is not None:
self._engine = create_engine(connection_string, connect_args={'auth_plugin': self._db_settings.auth_plugin})
if self._db_settings.encoding is not None:
self._engine.encoding = self._db_settings.encoding
if self._db_settings.case_sensitive is not None:
self._engine.case_sensitive = self._db_settings.case_sensitive
if self._db_settings.echo is not None:
self._engine.echo = self._db_settings.echo
self._engine.connect()
db_session = sessionmaker(bind=self._engine)
self._session = db_session()
Console.set_foreground_color(ForegroundColorEnum.green)
Console.write_line(f'[{__name__}] Connected to database')
Console.set_foreground_color(ForegroundColorEnum.default)
except Exception as e:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f'[{__name__}] Database connection failed -> {e}')
Console.set_foreground_color(ForegroundColorEnum.default)
exit()

30
src/cpl_core/database/connection/database_connection_abc.py Normal file
View File

@@ -0,0 +1,30 @@
from abc import abstractmethod, ABC
from sqlalchemy import engine
from sqlalchemy.orm import Session
class DatabaseConnectionABC(ABC):
r"""ABC for the :class:`cpl.database.connection.database_connection.DatabaseConnection`"""
@abstractmethod
def __init__(self): pass
@property
@abstractmethod
def engine(self) -> engine: pass
@property
@abstractmethod
def session(self) -> Session: pass
@abstractmethod
def connect(self, connection_string: str):
r"""Connects to a database by connection string
Parameter
---------
connection_string: :class:`str`
Database connection string, see: https://docs.sqlalchemy.org/en/14/core/engines.html
"""
pass

27
src/cpl_core/database/context/__init__.py Normal file
View File

@@ -0,0 +1,27 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.database.context'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .database_context import DatabaseContext
from .database_context_abc import DatabaseContextABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

56
src/cpl_core/database/context/database_context.py Normal file
View File

@@ -0,0 +1,56 @@
from sqlalchemy import engine, Table
from sqlalchemy.orm import Session
from cpl.console.console import Console
from cpl.console.foreground_color_enum import ForegroundColorEnum
from cpl.database.connection.database_connection import DatabaseConnection
from cpl.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.database.database_settings import DatabaseSettings
from cpl.database.database_model import DatabaseModel
class DatabaseContext(DatabaseContextABC):
r"""Representation of the database context
Parameter
---------
database_settings: :class:`cpl.database.database_settings.DatabaseSettings`
"""
def __init__(self, database_settings: DatabaseSettings):
DatabaseContextABC.__init__(self)
self._db: DatabaseConnectionABC = DatabaseConnection(database_settings)
self._tables: list[Table] = []
@property
def engine(self) -> engine:
return self._db.engine
@property
def session(self) -> Session:
return self._db.session
def connect(self, connection_string: str):
self._db.connect(connection_string)
self._create_tables()
def save_changes(self):
self._db.session.commit()
def _create_tables(self):
try:
for subclass in DatabaseModel.__subclasses__():
self._tables.append(subclass.__table__)
DatabaseModel.metadata.drop_all(self._db.engine, self._tables)
DatabaseModel.metadata.create_all(self._db.engine, self._tables, checkfirst=True)
Console.set_foreground_color(ForegroundColorEnum.green)
Console.write_line(f'[{__name__}] Created tables')
Console.set_foreground_color(ForegroundColorEnum.default)
except Exception as e:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f'[{__name__}] Creating tables failed -> {e}')
Console.set_foreground_color(ForegroundColorEnum.default)
exit()

40
src/cpl_core/database/context/database_context_abc.py Normal file
View File

@@ -0,0 +1,40 @@
from abc import abstractmethod, ABC
from sqlalchemy import engine
from sqlalchemy.orm import Session
class DatabaseContextABC(ABC):
r"""ABC for the :class:`cpl.database.context.database_context.DatabaseContext`"""
@abstractmethod
def __init__(self, *args):
pass
@property
@abstractmethod
def engine(self) -> engine: pass
@property
@abstractmethod
def session(self) -> Session: pass
@abstractmethod
def connect(self, connection_string: str):
r"""Connects to a database by connection string
Parameter
---------
connection_string: :class:`str`
Database connection string, see: https://docs.sqlalchemy.org/en/14/core/engines.html
"""
pass
def save_changes(self):
r"""Saves changes of the database"""
pass
@abstractmethod
def _create_tables(self):
r"""Create all tables for application from database model"""
pass

3
src/cpl_core/database/database_model.py Normal file
View File

@@ -0,0 +1,3 @@
from sqlalchemy.ext.declarative import declarative_base
DatabaseModel: declarative_base = declarative_base()

97
src/cpl_core/database/database_settings.py Normal file
View File

@@ -0,0 +1,97 @@
import traceback
from typing import Optional
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.console.console import Console
from cpl.console.foreground_color_enum import ForegroundColorEnum
from cpl.database.database_settings_name_enum import DatabaseSettingsNameEnum
class DatabaseSettings(ConfigurationModelABC):
r"""Represents settings for the database connection"""
def __init__(self):
ConfigurationModelABC.__init__(self)
self._auth_plugin: Optional[str] = None
self._connection_string: Optional[str] = None
self._credentials: Optional[str] = None
self._encoding: Optional[str] = None
self._case_sensitive: Optional[bool] = None
self._echo: Optional[bool] = None
@property
def auth_plugin(self) -> str:
return self._auth_plugin
@auth_plugin.setter
def auth_plugin(self, auth_plugin: str):
self._auth_plugin = auth_plugin
@property
def connection_string(self) -> str:
return self._connection_string
@connection_string.setter
def connection_string(self, connection_string: str):
self._connection_string = connection_string
@property
def credentials(self) -> str:
return self._credentials
@credentials.setter
def credentials(self, credentials: str):
self._credentials = credentials
@property
def encoding(self) -> str:
return self._encoding
@encoding.setter
def encoding(self, encoding: str) -> None:
self._encoding = encoding
@property
def case_sensitive(self) -> bool:
return self._case_sensitive
@case_sensitive.setter
def case_sensitive(self, case_sensitive: bool) -> None:
self._case_sensitive = case_sensitive
@property
def echo(self) -> bool:
return self._echo
@echo.setter
def echo(self, echo: bool) -> None:
self._echo = echo
def from_dict(self, settings: dict):
r"""Sets attributes from given dict
Parameter
---------
settings: :class:`dict`
"""
try:
self._connection_string = settings[DatabaseSettingsNameEnum.connection_string.value]
self._credentials = settings[DatabaseSettingsNameEnum.credentials.value]
if DatabaseSettingsNameEnum.auth_plugin.value in settings:
self._auth_plugin = settings[DatabaseSettingsNameEnum.auth_plugin.value]
if DatabaseSettingsNameEnum.encoding.value in settings:
self._encoding = settings[DatabaseSettingsNameEnum.encoding.value]
if DatabaseSettingsNameEnum.case_sensitive.value in settings:
self._case_sensitive = bool(settings[DatabaseSettingsNameEnum.case_sensitive.value])
if DatabaseSettingsNameEnum.echo.value in settings:
self._echo = bool(settings[DatabaseSettingsNameEnum.echo.value])
except Exception as e:
Console.set_foreground_color(ForegroundColorEnum.red)
Console.write_line(f'[ ERROR ] [ {__name__} ]: Reading error in {self.__name__} settings')
Console.write_line(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')
Console.set_foreground_color(ForegroundColorEnum.default)

11
src/cpl_core/database/database_settings_name_enum.py Normal file
View File

@@ -0,0 +1,11 @@
from enum import Enum
class DatabaseSettingsNameEnum(Enum):
connection_string = 'ConnectionString'
credentials = 'Credentials'
encoding = 'Encoding'
case_sensitive = 'CaseSensitive'
echo = 'Echo'
auth_plugin = 'AuthPlugin'

31
src/cpl_core/dependency_injection/__init__.py Normal file
View File

@@ -0,0 +1,31 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.dependency_injection'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .service_collection import ServiceCollection
from .service_collection_abc import ServiceCollectionABC
from .service_descriptor import ServiceDescriptor
from .service_lifetime_enum import ServiceLifetimeEnum
from .service_provider import ServiceProvider
from .service_provider_abc import ServiceProviderABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

71
src/cpl_core/dependency_injection/service_collection.py Normal file
View File

@@ -0,0 +1,71 @@
from typing import Union, Type, Callable, Optional
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.database.database_settings import DatabaseSettings
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.dependency_injection.service_provider import ServiceProvider
from cpl.logging.logger_service import Logger
from cpl.logging.logger_abc import LoggerABC
from cpl.utils.credential_manager import CredentialManager
class ServiceCollection(ServiceCollectionABC):
r"""Representation of the collection of services"""
def __init__(self, config: ConfigurationABC):
ServiceCollectionABC.__init__(self)
self._configuration: ConfigurationABC = config
self._database_context: Optional[DatabaseContextABC] = None
self._service_descriptors: list[ServiceDescriptor] = []
def _add_descriptor(self, service: Union[type, object], lifetime: ServiceLifetimeEnum):
found = False
for descriptor in self._service_descriptors:
if isinstance(service, descriptor.service_type):
found = True
if found:
service_type = service
if not isinstance(service, type):
service_type = type(service)
raise Exception(f'Service of type {service_type} already exists')
self._service_descriptors.append(ServiceDescriptor(service, lifetime))
def add_db_context(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
self._database_context = db_context_type(db_settings)
self._database_context.connect(CredentialManager.build_string(db_settings.connection_string, db_settings.credentials))
def add_logging(self):
self.add_singleton(LoggerABC, Logger)
def add_singleton(self, service_type: Union[type, object], service: Union[type, object] = None):
impl = None
if service is not None:
if isinstance(service, type):
impl = self.build_service_provider().build_service(service)
self._add_descriptor(impl, ServiceLifetimeEnum.singleton)
else:
if isinstance(service_type, type):
impl = self.build_service_provider().build_service(service_type)
self._add_descriptor(impl, ServiceLifetimeEnum.singleton)
def add_scoped(self, service_type: Type, service: Callable = None):
raise Exception('Not implemented')
def add_transient(self, service_type: Union[type], service: Union[type] = None):
if service is not None:
self._add_descriptor(service, ServiceLifetimeEnum.transient)
else:
self._add_descriptor(service_type, ServiceLifetimeEnum.transient)
def build_service_provider(self) -> ServiceProviderABC:
return ServiceProvider(self._service_descriptors, self._configuration, self._database_context)

82
src/cpl_core/dependency_injection/service_collection_abc.py Normal file
View File

@@ -0,0 +1,82 @@
from abc import abstractmethod, ABC
from collections import Callable
from typing import Type
from cpl.database.database_settings import DatabaseSettings
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
class ServiceCollectionABC(ABC):
r"""ABC for the class :class:`cpl.dependency_injection.service_collection.ServiceCollection`"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def add_db_context(self, db_context: Type[DatabaseContextABC], db_settings: DatabaseSettings):
r"""Adds database context
Parameter
---------
db_context: Type[:class:`cpl.database.context.database_context_abc.DatabaseContextABC`]
Database context
db_settings: :class:`cpl.database.database_settings.DatabaseSettings`
Database settings
"""
pass
@abstractmethod
def add_logging(self):
r"""Adds the CPL internal logger"""
pass
@abstractmethod
def add_transient(self, service_type: Type, service: Callable = None):
r"""Adds a service with transient lifetime
Parameter
---------
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
"""
pass
@abstractmethod
def add_scoped(self, service_type: Type, service: Callable = None):
r"""Adds a service with scoped lifetime
Parameter
---------
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
"""
pass
@abstractmethod
def add_singleton(self, service_type: Type, service: Callable = None):
r"""Adds a service with singleton lifetime
Parameter
---------
service_type: :class:`Type`
Type of the service
service: :class:`Callable`
Object of the service
"""
pass
@abstractmethod
def build_service_provider(self) -> ServiceProviderABC:
r"""Creates instance of the service provider
Returns
-------
Object of type :class:`cpl.dependency_injection.service_provider_abc.ServiceProviderABC`
"""
pass

42
src/cpl_core/dependency_injection/service_descriptor.py Normal file
View File

@@ -0,0 +1,42 @@
from typing import Union, Optional
from cpl.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
class ServiceDescriptor:
r"""Descriptor of a service
Parameter
---------
implementation: Union[:class:`type`, Optional[:class:`object`]]
Object or type of service
lifetime: :class:`cpl.dependency_injection.service_lifetime_enum.ServiceLifetimeEnum`
Lifetime of the service
"""
def __init__(self, implementation: Union[type, Optional[object]], lifetime: ServiceLifetimeEnum):
self._service_type = implementation
self._implementation = implementation
self._lifetime = lifetime
if not isinstance(implementation, type):
self._service_type = type(implementation)
else:
self._implementation = None
@property
def service_type(self) -> type:
return self._service_type
@property
def implementation(self) -> Union[type, Optional[object]]:
return self._implementation
@implementation.setter
def implementation(self, implementation: Union[type, Optional[object]]):
self._implementation = implementation
@property
def lifetime(self) -> ServiceLifetimeEnum:
return self._lifetime

8
src/cpl_core/dependency_injection/service_lifetime_enum.py Normal file
View File

@@ -0,0 +1,8 @@
from enum import Enum
class ServiceLifetimeEnum(Enum):
singleton = 0
scoped = 1 # not supported yet
transient = 2

101
src/cpl_core/dependency_injection/service_provider.py Normal file
View File

@@ -0,0 +1,101 @@
from collections import Callable
from inspect import signature, Parameter
from typing import Optional
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
class ServiceProvider(ServiceProviderABC):
r"""Provider for the services
Parameter
---------
service_descriptors: list[:class:`cpl.dependency_injection.service_descriptor.ServiceDescriptor`]
Descriptor of the service
config: :class:`cpl.configuration.configuration_abc.ConfigurationABC`
CPL Configuration
db_context: Optional[:class:`cpl.database.context.database_context_abc.DatabaseContextABC`]
Database representation
"""
def __init__(self, service_descriptors: list[ServiceDescriptor], config: ConfigurationABC, db_context: Optional[DatabaseContextABC]):
ServiceProviderABC.__init__(self)
self._service_descriptors: list[ServiceDescriptor] = service_descriptors
self._configuration: ConfigurationABC = config
self._database_context = db_context
def _find_service(self, service_type: type) -> [ServiceDescriptor]:
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
return descriptor
return None
def _get_service(self, parameter: Parameter) -> object:
for descriptor in self._service_descriptors:
if descriptor.service_type == parameter.annotation or issubclass(descriptor.service_type, parameter.annotation):
if descriptor.implementation is not None:
return descriptor.implementation
implementation = self.build_service(descriptor.service_type)
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptor.implementation = implementation
return implementation
def build_service(self, service_type: type) -> object:
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
if descriptor.implementation is not None:
service_type = type(descriptor.implementation)
else:
service_type = descriptor.service_type
break
sig = signature(service_type.__init__)
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
elif issubclass(parameter.annotation, ApplicationEnvironmentABC):
params.append(self._configuration.environment)
elif issubclass(parameter.annotation, DatabaseContextABC):
params.append(self._database_context)
elif issubclass(parameter.annotation, ConfigurationModelABC):
params.append(self._configuration.get_configuration(parameter.annotation))
elif issubclass(parameter.annotation, ConfigurationABC):
params.append(self._configuration)
else:
params.append(self._get_service(parameter))
return service_type(*params)
def get_service(self, service_type: type) -> Optional[Callable[object]]:
result = self._find_service(service_type)
if result is None:
return None
if result.implementation is not None:
return result.implementation
implementation = self.build_service(service_type)
if result.lifetime == ServiceLifetimeEnum.singleton:
result.implementation = implementation
return implementation

41
src/cpl_core/dependency_injection/service_provider_abc.py Normal file
View File

@@ -0,0 +1,41 @@
from abc import abstractmethod, ABC
from collections import Callable
from typing import Type, Optional
class ServiceProviderABC(ABC):
r"""ABC for the class :class:`cpl.dependency_injection.service_provider.ServiceProvider`"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def build_service(self, service_type: Type) -> object:
r"""Creates instance of given type
Parameter
---------
instance_type: :class:`Type`
The type of the searched instance
Returns
-------
Object of the given type
"""
pass
@abstractmethod
def get_service(self, instance_type: Type) -> Optional[Callable[object]]:
r"""Returns instance of given type
Parameter
---------
instance_type: :class:`Type`
The type of the searched instance
Returns
-------
Object of type Optional[Callable[:class:`object`]]
"""
pass

28
src/cpl_core/environment/__init__.py Normal file
View File

@@ -0,0 +1,28 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.environment'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
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='2021', minor='10', micro='6')

96
src/cpl_core/environment/application_environment.py Normal file
View File

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

88
src/cpl_core/environment/application_environment_abc.py Normal file
View File

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

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

@@ -0,0 +1,9 @@
from enum import Enum
class EnvironmentNameEnum(Enum):
production = 'production'
staging = 'staging'
testing = 'testing'
development = 'development'

30
src/cpl_core/logging/__init__.py Normal file
View File

@@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.logging'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .logger_service import Logger
from .logger_abc import LoggerABC
from .logging_level_enum import LoggingLevelEnum
from .logging_settings import LoggingSettings
from .logging_settings_name_enum import LoggingSettingsNameEnum
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

102
src/cpl_core/logging/logger_abc.py Normal file
View File

@@ -0,0 +1,102 @@
from abc import abstractmethod, ABC
class LoggerABC(ABC):
r"""ABC for :class:`cpl.logging.logger_service.Logger`"""
@abstractmethod
def __init__(self):
ABC.__init__(self)
@abstractmethod
def header(self, string: str):
r"""Writes a header message
Parameter
---------
string: :class:`str`
String to write as header
"""
pass
@abstractmethod
def trace(self, name: str, message: str):
r"""Writes a trace message
Parameter
---------
name: :class:`str`
Message name
message: :class:`str`
Message string
"""
pass
@abstractmethod
def debug(self, name: str, message: str):
r"""Writes a debug message
Parameter
---------
name: :class:`str`
Message name
message: :class:`str`
Message string
"""
pass
@abstractmethod
def info(self, name: str, message: str):
r"""Writes an information
Parameter
---------
name: :class:`str`
Message name
message: :class:`str`
Message string
"""
pass
@abstractmethod
def warn(self, name: str, message: str):
r"""Writes an warning
Parameter
---------
name: :class:`str`
Message name
message: :class:`str`
Message string
"""
pass
@abstractmethod
def error(self, name: str, message: str, ex: Exception = None):
r"""Writes an error
Parameter
---------
name: :class:`str`
Error name
message: :class:`str`
Error message
ex: :class:`Exception`
Thrown exception
"""
pass
@abstractmethod
def fatal(self, name: str, message: str, ex: Exception = None):
r"""Writes an error and ends the program
Parameter
---------
name: :class:`str`
Error name
message: :class:`str`
Error message
ex: :class:`Exception`
Thrown exception
"""
pass

256
src/cpl_core/logging/logger_service.py Normal file
View File

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

12
src/cpl_core/logging/logging_level_enum.py Normal file
View File

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

63
src/cpl_core/logging/logging_settings.py Normal file
View File

@@ -0,0 +1,63 @@
import traceback
from typing import Optional
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.console.console import Console
from cpl.console.foreground_color_enum import ForegroundColorEnum
from cpl.logging.logging_level_enum import LoggingLevelEnum
from cpl.logging.logging_settings_name_enum import LoggingSettingsNameEnum
class LoggingSettings(ConfigurationModelABC):
r"""Representation of logging settings"""
def __init__(self):
ConfigurationModelABC.__init__(self)
self._path: Optional[str] = None
self._filename: Optional[str] = None
self._console: Optional[LoggingLevelEnum] = None
self._level: Optional[LoggingLevelEnum] = None
@property
def path(self) -> str:
return self._path
@path.setter
def path(self, path: str) -> None:
self._path = path
@property
def filename(self) -> str:
return self._filename
@filename.setter
def filename(self, filename: str) -> None:
self._filename = filename
@property
def console(self) -> LoggingLevelEnum:
return self._console
@console.setter
def console(self, console: LoggingLevelEnum) -> None:
self._console = console
@property
def level(self) -> LoggingLevelEnum:
return self._level
@level.setter
def level(self, level: LoggingLevelEnum) -> None:
self._level = level
def from_dict(self, settings: dict):
try:
self._path = settings[LoggingSettingsNameEnum.path.value]
self._filename = settings[LoggingSettingsNameEnum.filename.value]
self._console = LoggingLevelEnum[settings[LoggingSettingsNameEnum.console_level.value]]
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 {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 Normal file
View File

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

30
src/cpl_core/mailing/__init__.py Normal file
View File

@@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.mailing'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .email import EMail
from .email_client_service import EMailClient
from .email_client_abc import EMailClientABC
from .email_client_settings import EMailClientSettings
from .email_client_settings_name_enum import EMailClientSettingsNameEnum
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

140
src/cpl_core/mailing/email.py Normal file
View File

@@ -0,0 +1,140 @@
import re
class EMail:
r"""Represents an email
Parameter
---------
header: list[:class:`str`]
Header of the E-Mail
subject: :class:`str`
Subject of the E-Mail
body: :class:`str`
Body of the E-Mail
transceiver: :class:`str`
Transceiver of the E-Mail
receiver: list[:class:`str`]
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):
self._header: list[str] = header
self._subject: str = subject
self._body: str = body
self._transceiver: str = transceiver
self._receiver: list[str] = receiver
@property
def header(self) -> str:
return '\r\n'.join(self._header)
@property
def header_list(self) -> list[str]:
return self._header
@header.setter
def header(self, header: list[str]):
self._header = header
@property
def subject(self) -> str:
return self._subject
@subject.setter
def subject(self, subject: str):
self._subject = subject
@property
def body(self) -> str:
return self._body
@body.setter
def body(self, body: str):
self._body = body
@property
def transceiver(self) -> str:
return self._transceiver
@transceiver.setter
def transceiver(self, transceiver: str):
if self.check_mail(transceiver):
self._transceiver = transceiver
else:
raise Exception(f'Invalid email: {transceiver}')
@property
def receiver(self) -> str:
return ','.join(self._receiver)
@property
def receiver_list(self) -> list[str]:
return self._receiver
@receiver.setter
def receiver(self, receiver: list[str]):
self._receiver = receiver
@staticmethod
def check_mail(address: str) -> bool:
r"""Checks if an email is valid
Parameter
---------
address: :class:`str`
The address to check
Returns
-------
Result if E-Mail is valid or not
"""
return bool(re.search('^\\w+([.-]?\\w+)*@\\w+([.-]?\\w+)*(.\\w{2,3})+$', address))
def add_header(self, header: str):
r"""Adds header
Parameter
---------
header: :class:`str`
The header of the E-Mail
"""
if self._header is None:
self._header = []
self._header.append(header)
def add_receiver(self, receiver: str):
r"""Adds receiver
Parameter
---------
receiver: :class:`str`
The receiver of the E-Mail
"""
if self._receiver is None:
self._receiver = []
if self.check_mail(receiver):
self._receiver.append(receiver)
else:
raise Exception(f'Invalid email: {receiver}')
def get_content(self, transceiver: str):
r"""Returns the mail as string
Parameter
---------
transceiver: :class:`str`
The transceiver of the E-Mail
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')

27
src/cpl_core/mailing/email_client_abc.py Normal file
View File

@@ -0,0 +1,27 @@
from abc import abstractmethod, ABC
from cpl.mailing.email import EMail
class EMailClientABC(ABC):
"""ABC of :class:`cpl.mailing.email_client_service.EMailClient`"""
@abstractmethod
def __init__(self):
ABC.__init__(self)
@abstractmethod
def connect(self):
r"""Connects to server"""
pass
@abstractmethod
def send_mail(self, email: EMail):
r"""Sends email
Parameter
---------
email: :class:`cpl.mailing.email.EMail`
Object of the E-Mail to send
"""
pass

83
src/cpl_core/mailing/email_client_service.py Normal file
View File

@@ -0,0 +1,83 @@
import ssl
from smtplib import SMTP
from typing import Optional
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.logging.logger_abc import LoggerABC
from cpl.mailing.email import EMail
from cpl.mailing.email_client_abc import EMailClientABC
from cpl.mailing.email_client_settings import EMailClientSettings
from cpl.utils.credential_manager import CredentialManager
class EMailClient(EMailClientABC):
r"""Service to send emails
Parameter
---------
environment: :class:`cpl.environment.application_environment_abc.ApplicationEnvironmentABC`
Environment of the application
logger: :class:`cpl.logging.logger_abc.LoggerABC`
The logger to use
mail_settings: :class:`cpl.mailing.email_client_settings.EMailClientSettings`
Settings for mailing
"""
def __init__(self, environment: ApplicationEnvironmentABC, logger: LoggerABC, mail_settings: EMailClientSettings):
EMailClientABC.__init__(self)
self._environment = environment
self._mail_settings = mail_settings
self._logger = logger
self._server: Optional[SMTP] = None
self.create()
def create(self):
r"""Creates connection"""
self._logger.trace(__name__, f'Started {__name__}.create')
self.connect()
self._logger.trace(__name__, f'Stopped {__name__}.create')
def connect(self):
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._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.debug(__name__, 'Try to start tls')
self._server.starttls(context=ssl.create_default_context())
self._logger.info(__name__, 'Started tls')
except Exception as e:
self._logger.error(__name__, 'Cannot connect to mail server', e)
self._logger.trace(__name__, f'Stopped {__name__}.connect')
def login(self):
r"""Login to server"""
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}')
except Exception as e:
self._logger.error(__name__, 'Cannot login to mail server', e)
self._logger.trace(__name__, f'Stopped {__name__}.login')
def send_mail(self, email: EMail):
self._logger.trace(__name__, f'Started {__name__}.send_mail')
try:
self.login()
email.body += f'\n\nDies ist eine automatische E-Mail.' \
f'\nGesendet von {self._environment.application_name}-{self._environment.environment_name}@{self._environment.host_name} für ' \
f'{self._environment.customer}.'
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')

60
src/cpl_core/mailing/email_client_settings.py Normal file
View File

@@ -0,0 +1,60 @@
import traceback
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.console.console import Console
from cpl.mailing.email_client_settings_name_enum import EMailClientSettingsNameEnum
class EMailClientSettings(ConfigurationModelABC):
r"""Representation of mailing settings"""
def __init__(self):
ConfigurationModelABC.__init__(self)
self._host: str = ''
self._port: int = 0
self._user_name: str = ''
self._credentials: str = ''
@property
def host(self) -> str:
return self._host
@host.setter
def host(self, host: str) -> None:
self._host = host
@property
def port(self) -> int:
return self._port
@port.setter
def port(self, port: int) -> None:
self._port = port
@property
def user_name(self) -> str:
return self._user_name
@user_name.setter
def user_name(self, user_name: str) -> None:
self._user_name = user_name
@property
def credentials(self) -> str:
return self._credentials
@credentials.setter
def credentials(self, credentials: str) -> None:
self._credentials = credentials
def from_dict(self, settings: dict):
try:
self._host = settings[EMailClientSettingsNameEnum.host.value]
self._port = settings[EMailClientSettingsNameEnum.port.value]
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 {self.__name__} settings')
Console.error(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}')

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

@@ -0,0 +1,9 @@
from enum import Enum
class EMailClientSettingsNameEnum(Enum):
host = 'Host'
port = 'Port'
user_name = 'UserName'
credentials = 'Credentials'

27
src/cpl_core/time/__init__.py Normal file
View File

@@ -0,0 +1,27 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.time'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .time_format_settings import TimeFormatSettings
from .time_format_settings_names_enum import TimeFormatSettingsNamesEnum
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

62
src/cpl_core/time/time_format_settings.py Normal file
View File

@@ -0,0 +1,62 @@
import traceback
from typing import Optional
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.console.console import Console
from cpl.console.foreground_color_enum import ForegroundColorEnum
from cpl.time.time_format_settings_names_enum import TimeFormatSettingsNamesEnum
class TimeFormatSettings(ConfigurationModelABC):
r"""Representation of time format settings"""
def __init__(self):
ConfigurationModelABC.__init__(self)
self._date_format: Optional[str] = None
self._time_format: Optional[str] = None
self._date_time_format: Optional[str] = None
self._date_time_log_format: Optional[str] = None
@property
def date_format(self) -> str:
return self._date_format
@date_format.setter
def date_format(self, date_format: str) -> None:
self._date_format = date_format
@property
def time_format(self) -> str:
return self._time_format
@time_format.setter
def time_format(self, time_format: str):
self._time_format = time_format
@property
def date_time_format(self) -> str:
return self._date_time_format
@date_time_format.setter
def date_time_format(self, date_time_format: str) -> None:
self._date_time_format = date_time_format
@property
def date_time_log_format(self):
return self._date_time_log_format
@date_time_log_format.setter
def date_time_log_format(self, date_time_now_format: str) -> None:
self._date_time_log_format = date_time_now_format
def from_dict(self, settings: dict):
try:
self._date_format = settings[TimeFormatSettingsNamesEnum.date_format.value]
self._time_format = settings[TimeFormatSettingsNamesEnum.time_format.value]
self._date_time_format = settings[TimeFormatSettingsNamesEnum.date_time_format.value]
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 {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 Normal file
View File

@@ -0,0 +1,9 @@
from enum import Enum
class TimeFormatSettingsNamesEnum(Enum):
date_format = 'DateFormat'
time_format = 'TimeFormat'
date_time_format = 'DateTimeFormat'
date_time_log_format = 'DateTimeLogFormat'

28
src/cpl_core/utils/__init__.py Normal file
View File

@@ -0,0 +1,28 @@
# -*- coding: utf-8 -*-
"""
sh_cpl-core sh-edraft Common Python library
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library
:copyright: (c) 2020 - 2021 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_core.utils'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 - 2021 sh-edraft.de'
__version__ = '2021.10.6'
from collections import namedtuple
# imports:
from .credential_manager import CredentialManager
from .string import String
from .pip import Pip
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2021', minor='10', micro='6')

53
src/cpl_core/utils/credential_manager.py Normal file
View File

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

151
src/cpl_core/utils/pip.py Normal file
View File

@@ -0,0 +1,151 @@
import os
import subprocess
import sys
from contextlib import suppress
from typing import Optional
class Pip:
r"""Executes pip commands"""
_executable = sys.executable
_env = os.environ
_is_venv = False
"""Getter"""
@classmethod
def get_executable(cls) -> str:
return cls._executable
"""Setter"""
@classmethod
def set_executable(cls, executable: str):
r"""Sets the executable
Parameter
---------
executable: :class:`str`
The python command
"""
if executable is not None and executable != sys.executable:
cls._executable = executable
if os.path.islink(cls._executable):
cls._is_venv = True
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', '')
else:
cls._env['PATH'] = f'{path}/bin' + os.pathsep + os.environ.get('PATH', '')
cls._env['VIRTUAL_ENV'] = path
@classmethod
def reset_executable(cls):
r"""Resets the executable to system standard"""
cls._executable = sys.executable
cls._is_venv = False
"""Public utils functions"""
@classmethod
def get_package(cls, package: str) -> Optional[str]:
r"""Gets given package py local pip list
Parameter
---------
package: :class:`str`
Returns
-------
The package name as string
"""
result = None
with suppress(Exception):
args = [cls._executable, "-m", "pip", "show", package]
if cls._is_venv:
args = ["pip", "show", package]
result = subprocess.check_output(
args,
stderr=subprocess.DEVNULL, env=cls._env
)
if result is None:
return None
new_package: list[str] = str(result, 'utf-8').lower().split('\n')
new_version = ''
for atr in new_package:
if 'version' in atr:
new_version = atr.split(': ')[1]
if new_version != '':
return f'{package}=={new_version}'
return package
@classmethod
def get_outdated(cls) -> bytes:
r"""Gets table of outdated packages
Returns
-------
Bytes string of the command result
"""
args = [cls._executable, "-m", "pip", "list", "--outdated"]
if cls._is_venv:
args = ["pip", "list", "--outdated"]
return subprocess.check_output(args, env=cls._env)
@classmethod
def install(cls, package: str, *args, source: str = None, stdout=None, stderr=None):
r"""Installs given package
Parameter
---------
package: :class:`str`
The name of the package
args: :class:`list`
Arguments for the command
source: :class:`str`
Extra index URL
stdout: :class:`str`
Stdout of subprocess.run
stderr: :class:`str`
Stderr of subprocess.run
"""
pip_args = [cls._executable, "-m", "pip", "install"]
if cls._is_venv:
pip_args = ["pip", "install"]
for arg in args:
pip_args.append(arg)
if source is not None:
pip_args.append(f'--extra-index-url')
pip_args.append(source)
pip_args.append(package)
subprocess.run(pip_args, stdout=stdout, stderr=stderr, env=cls._env)
@classmethod
def uninstall(cls, package: str, stdout=None, stderr=None):
r"""Uninstalls given package
Parameter
---------
package: :class:`str`
The name of the package
stdout: :class:`str`
Stdout of subprocess.run
stderr: :class:`str`
Stderr of subprocess.run
"""
args = [cls._executable, "-m", "pip", "uninstall", "--yes", package]
if cls._is_venv:
args = ["pip", "uninstall", "--yes", package]
subprocess.run(
args,
stdout=stdout, stderr=stderr, env=cls._env
)

91
src/cpl_core/utils/string.py Normal file
View File

@@ -0,0 +1,91 @@
import re
import string
import random
class String:
r"""Useful functions for strings"""
@staticmethod
def convert_to_camel_case(chars: str) -> str:
r"""Converts string to camel case
Parameter
---------
chars: :class:`str`
String to convert
Returns
-------
String converted to CamelCase
"""
converted_name = chars
char_set = string.punctuation + ' '
for char in char_set:
if char in converted_name:
converted_name = ''.join(word.title() for word in converted_name.split(char))
return converted_name
@staticmethod
def convert_to_snake_case(chars: str) -> str:
r"""Converts string to snake case
Parameter
---------
chars: :class:`str`
String to convert
Returns
-------
String converted to snake_case
"""
# convert to train-case to CamelCase
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()
@staticmethod
def first_to_upper(chars: str) -> str:
r"""Converts first char to upper
Parameter
---------
chars: :class:`str`
String to convert
Returns
-------
String with first char as upper
"""
return f'{chars[0].upper()}{chars[1:]}'
@staticmethod
def first_to_lower(chars: str) -> str:
r"""Converts first char to lower
Parameter
---------
chars: :class:`str`
String to convert
Returns
-------
String with first char as lower
"""
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
-------
String of random chars
"""
return ''.join(random.choice(chars) for _ in range(length))