Refactored code

This commit is contained in:
2021-03-03 10:47:52 +01:00
parent a7c2946ba5
commit 68c136a16f
205 changed files with 2207 additions and 1010 deletions

0
src/cpl/__init__.py Normal file
View File

View File

View File

@@ -0,0 +1,32 @@
from abc import ABC, abstractmethod
from typing import Type, Optional
from cpl.application.application_host_abc import ApplicationHostABC
from cpl.application.startup_abc import StartupABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.dependency_injection.service_provider_base import ServiceProviderABC
class ApplicationABC(ABC):
@abstractmethod
def __init__(self):
self._startup: Optional[StartupABC] = None
self._app_host: Optional[ApplicationHostABC] = None
self._services: Optional[ServiceProviderABC] = None
self._configuration: Optional[ConfigurationABC] = None
def use_startup(self, startup: Type[StartupABC]):
self._startup = startup()
def build(self):
if self._startup is None:
print('Startup is empty')
exit()
self._app_host = self._startup.create_application_host()
self._configuration = self._startup.create_configuration()
self._services = self._startup.create_services()
@abstractmethod
def run(self): pass

View File

@@ -0,0 +1,42 @@
import atexit
from datetime import datetime
from cpl.application.application_host_abc import ApplicationHostABC
from cpl.application.application_runtime import ApplicationRuntime
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration import Configuration
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.console.console import Console
from cpl.dependency_injection.service_provider import ServiceProvider
from cpl.dependency_injection.service_provider_base import ServiceProviderABC
class ApplicationHost(ApplicationHostABC):
def __init__(self):
ApplicationHostABC.__init__(self)
# Init
self._config = Configuration()
self._app_runtime = ApplicationRuntime(self._config)
self._services = ServiceProvider(self._app_runtime)
# Set vars
self._start_time: datetime = datetime.now()
self._end_time: datetime = datetime.now()
atexit.register(Console.close)
@property
def configuration(self) -> ConfigurationABC:
return self._config
@property
def application_runtime(self) -> ApplicationRuntimeABC:
return self._app_runtime
@property
def services(self) -> ServiceProviderABC:
return self._services
def create(self): pass

View File

@@ -0,0 +1,23 @@
from abc import ABC, abstractmethod
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.dependency_injection.service_provider_base import ServiceProviderABC
class ApplicationHostABC(ABC):
@abstractmethod
def __init__(self): pass
@property
@abstractmethod
def configuration(self) -> ConfigurationABC: pass
@property
@abstractmethod
def application_runtime(self) -> ApplicationRuntimeABC: pass
@property
@abstractmethod
def services(self) -> ServiceProviderABC: pass

View File

@@ -0,0 +1,38 @@
from datetime import datetime
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
class ApplicationRuntime(ApplicationRuntimeABC):
def __init__(self, config: ConfigurationABC):
ApplicationRuntimeABC.__init__(self)
self._app_configuration = config
self._start_time: datetime = datetime.now()
self._end_time: datetime = datetime.now()
@property
def configuration(self) -> ConfigurationABC:
return self._app_configuration
@property
def start_time(self) -> datetime:
return self._start_time
@start_time.setter
def start_time(self, start_time: datetime):
self._start_time = 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()

View File

@@ -0,0 +1,34 @@
from abc import ABC, abstractmethod
from datetime import datetime
from cpl.configuration.configuration_abc import ConfigurationABC
class ApplicationRuntimeABC(ABC):
@abstractmethod
def __init__(self): pass
@property
@abstractmethod
def configuration(self) -> ConfigurationABC: 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

View File

@@ -0,0 +1,20 @@
from abc import ABC, abstractmethod
from cpl.application.application_host_abc import ApplicationHostABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.dependency_injection.service_provider_base import ServiceProviderABC
class StartupABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def create_application_host(self) -> ApplicationHostABC: pass
@abstractmethod
def create_configuration(self) -> ConfigurationABC: pass
@abstractmethod
def create_services(self) -> ServiceProviderABC: pass

View File

View File

@@ -0,0 +1,118 @@
import json
import os
import sys
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.configuration.configuration_variable_name import ConfigurationVariableName
from cpl.console.console import Console
from cpl.console.foreground_color import ForegroundColor
from cpl.environment.hosting_environment import HostingEnvironment
from cpl.environment.environment_abc import EnvironmentABC
from cpl.environment.environment_name import EnvironmentName
class Configuration(ConfigurationABC):
def __init__(self):
ConfigurationABC.__init__(self)
self._hosting_environment = HostingEnvironment()
self._config: dict[type, ConfigurationModelABC] = {}
@property
def environment(self) -> EnvironmentABC:
return self._hosting_environment
@staticmethod
def _print_info(name: str, message: str):
Console.set_foreground_color(ForegroundColor.green)
Console.write_line(f'[{name}] {message}')
Console.set_foreground_color(ForegroundColor.default)
@staticmethod
def _print_warn(name: str, message: str):
Console.set_foreground_color(ForegroundColor.yellow)
Console.write_line(f'[{name}] {message}')
Console.set_foreground_color(ForegroundColor.default)
@staticmethod
def _print_error(name: str, message: str):
Console.set_foreground_color(ForegroundColor.red)
Console.write_line(f'[{name}] {message}')
Console.set_foreground_color(ForegroundColor.default)
def _set_variable(self, name: str, value: str):
if name == ConfigurationVariableName.environment.value:
self._hosting_environment.environment_name = EnvironmentName(value)
elif name == ConfigurationVariableName.name.value:
self._hosting_environment.application_name = value
elif name == ConfigurationVariableName.customer.value:
self._hosting_environment.customer = value
def add_environment_variables(self, prefix: str):
for variable in ConfigurationVariableName.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_argument_variables(self):
for arg in sys.argv[1:]:
try:
argument = arg.split('--')[1].split('=')[0].upper()
value = arg.split('=')[1]
if argument not in ConfigurationVariableName.to_list():
raise Exception(f'Invalid argument name: {argument}')
self._set_variable(argument, value)
except Exception as e:
self._print_error(__name__, f'Invalid argument: {arg} -> {e}')
exit()
def add_json_file(self, name: str, optional: bool = None):
if self._hosting_environment.content_root_path.endswith('/') and not name.startswith('/'):
file_path = f'{self._hosting_environment.content_root_path}{name}'
else:
file_path = f'{self._hosting_environment.content_root_path}/{name}'
if not os.path.isfile(file_path):
if not optional:
self._print_error(__name__, f'File not found: {file_path}')
exit()
self._print_warn(__name__, f'Not Loaded config file: {file_path}')
return None
config_from_file = self._load_json_file(file_path)
for sub in ConfigurationModelABC.__subclasses__():
for key, value in config_from_file.items():
if sub.__name__ == key:
configuration = sub()
configuration.from_dict(value)
self.add_configuration(sub, configuration)
def _load_json_file(self, file: str) -> dict:
try:
# open config file, create if not exists
with open(file, encoding='utf-8') as cfg:
# load json
json_cfg = json.load(cfg)
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_configuration(self, key_type: type, value: ConfigurationModelABC):
self._config[key_type] = value
def get_configuration(self, search_type: type) -> ConfigurationModelABC:
if search_type not in self._config:
raise Exception(f'Config model by type {search_type} not found')
for config_model in self._config:
if config_model == search_type:
return self._config[config_model]

View File

@@ -0,0 +1,31 @@
from abc import abstractmethod, ABC
from collections import Callable
from typing import Type
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.environment.environment_abc import EnvironmentABC
class ConfigurationABC(ABC):
@abstractmethod
def __init__(self): pass
@property
@abstractmethod
def environment(self) -> EnvironmentABC: pass
@abstractmethod
def add_environment_variables(self, prefix: str): pass
@abstractmethod
def add_argument_variables(self): pass
@abstractmethod
def add_json_file(self, name: str, optional: bool = None): pass
@abstractmethod
def add_configuration(self, key_type: type, value: object): pass
@abstractmethod
def get_configuration(self, search_type: Type[ConfigurationModelABC]) -> Callable[ConfigurationModelABC]: pass

View File

@@ -0,0 +1,10 @@
from abc import ABC, abstractmethod
class ConfigurationModelABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def from_dict(self, settings: dict): pass

View File

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

View File

View File

@@ -0,0 +1,14 @@
from enum import Enum
class BackgroundColor(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'

205
src/cpl/console/console.py Normal file
View File

@@ -0,0 +1,205 @@
import os
from typing import Union, Optional
import pyfiglet
from tabulate import tabulate
from termcolor import colored
from cpl.console.background_color import BackgroundColor
from cpl.console.foreground_color import ForegroundColor
class Console:
_is_first_write = True
_background_color: BackgroundColor = BackgroundColor.default
_foreground_color: ForegroundColor = ForegroundColor.default
_x: Optional[int] = None
_y: Optional[int] = None
_disabled: bool = False
"""
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_background_color(cls, color: Union[BackgroundColor, str]):
if type(color) is str:
cls._background_color = BackgroundColor[color]
else:
cls._background_color = color
@classmethod
def set_foreground_color(cls, color: Union[ForegroundColor, str]):
if type(color) is str:
cls._foreground_color = ForegroundColor[color]
else:
cls._foreground_color = color
@classmethod
def reset_cursor_position(cls):
cls._x = None
cls._y = None
@classmethod
def set_cursor_position(cls, x: int, y: int):
cls._x = x
cls._y = y
"""
Useful protected methods
"""
@classmethod
def _output(cls, string: str, x: int = None, y: int = None, end='\n'):
if cls._is_first_write:
cls._is_first_write = False
args = []
colored_args = []
if x is not None and y is not None:
args.append(f'\033[{x};{y}H')
elif cls._x is not None and cls._y is not None:
args.append(f'\033[{cls._x};{cls._y}H')
colored_args.append(string)
if cls._foreground_color != ForegroundColor.default and cls._background_color == BackgroundColor.default:
colored_args.append(cls._foreground_color.value)
elif cls._foreground_color == ForegroundColor.default and cls._background_color != BackgroundColor.default:
colored_args.append(cls._background_color.value)
elif cls._foreground_color != ForegroundColor.default and cls._background_color != BackgroundColor.default:
colored_args.append(cls._foreground_color.value)
colored_args.append(cls._background_color.value)
args.append(colored(*colored_args))
print(*args, end=end)
"""
Useful public methods
"""
@classmethod
def banner(cls, string: str):
if cls._disabled:
return
ascii_banner = pyfiglet.figlet_format(string)
cls.write_line(ascii_banner)
@classmethod
def clear(cls):
os.system('cls' if os.name == 'nt' else 'clear')
@classmethod
def close(cls):
if cls._disabled:
return
Console.reset()
Console.write('\n\n\nPress any key to continue...')
Console.read_line()
exit()
@classmethod
def disable(cls):
cls._disabled = True
@classmethod
def error(cls, string: str, tb: str = None):
if cls._disabled:
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):
cls._disabled = False
@classmethod
def read(cls, output: str = None) -> str:
if output is not None:
cls.write(output)
return input()[0]
@classmethod
def read_line(cls, output: str = None) -> str:
if cls._disabled:
return ''
if output is not None:
cls.write(output)
return input()
@classmethod
def reset(cls):
cls._background_color = BackgroundColor.default
cls._foreground_color = ForegroundColor.default
@classmethod
def table(cls, header: list[str], values: list[list[str]]):
if cls._disabled:
return
table = tabulate(values, headers=header)
Console.write_line(table)
Console.write('\n')
@classmethod
def write(cls, *args):
if cls._disabled:
return
string = ' '.join(map(str, args))
cls._output(string, end='')
@classmethod
def write_at(cls, x: int, y: int, *args):
if cls._disabled:
return
string = ' '.join(map(str, args))
cls._output(string, x, y, end='')
@classmethod
def write_line(cls, *args):
if cls._disabled:
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):
if cls._disabled:
return
string = ' '.join(map(str, args))
if not cls._is_first_write:
cls._output('', end='')
cls._output(string, x, y, end='')

View File

@@ -0,0 +1,14 @@
from enum import Enum
class ForegroundColor(Enum):
default = 'default'
grey = 'grey'
red = 'red'
green = 'green'
yellow = 'yellow'
blue = 'blue'
magenta = 'magenta'
cyan = 'cyan'
white = 'white'

View File

View File

View File

@@ -0,0 +1,58 @@
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 import ForegroundColor
from cpl.database.connection.database_connection_abc import DatabaseConnectionABC
from cpl.database.database_settings import DatabaseSettings
class DatabaseConnection(DatabaseConnectionABC):
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(ForegroundColor.green)
Console.write_line(f'[{__name__}] Connected to database')
Console.set_foreground_color(ForegroundColor.default)
except Exception as e:
Console.set_foreground_color(ForegroundColor.red)
Console.write_line(f'[{__name__}] Database connection failed -> {e}')
Console.set_foreground_color(ForegroundColor.default)
exit()

View File

@@ -0,0 +1,21 @@
from abc import abstractmethod, ABC
from sqlalchemy import engine
from sqlalchemy.orm import Session
class DatabaseConnectionABC(ABC):
@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): pass

View File

View File

@@ -0,0 +1,50 @@
from sqlalchemy import engine, Table
from sqlalchemy.orm import Session
from cpl.console.console import Console
from cpl.console.foreground_color import ForegroundColor
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):
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 create(self):
pass
def connect(self, connection_string: str):
self._db.connect(connection_string)
self._create_tables()
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(ForegroundColor.green)
Console.write_line(f'[{__name__}] Created tables')
Console.set_foreground_color(ForegroundColor.default)
except Exception as e:
Console.set_foreground_color(ForegroundColor.red)
Console.write_line(f'[{__name__}] Creating tables failed -> {e}')
Console.set_foreground_color(ForegroundColor.default)
exit()

View File

@@ -0,0 +1,21 @@
from abc import abstractmethod, ABC
from sqlalchemy import engine
from sqlalchemy.orm import Session
class DatabaseContextABC(ABC):
@property
@abstractmethod
def engine(self) -> engine: pass
@property
@abstractmethod
def session(self) -> Session: pass
@abstractmethod
def connect(self, connection_string: str): pass
@abstractmethod
def _create_tables(self): pass

View File

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

View File

@@ -0,0 +1,90 @@
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 import ForegroundColor
from cpl.database.database_settings_name import DatabaseSettingsName
class DatabaseSettings(ConfigurationModelABC):
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):
try:
self._connection_string = settings[DatabaseSettingsName.connection_string.value]
self._credentials = settings[DatabaseSettingsName.credentials.value]
if DatabaseSettingsName.auth_plugin.value in settings:
self._auth_plugin = settings[DatabaseSettingsName.auth_plugin.value]
if DatabaseSettingsName.encoding.value in settings:
self._encoding = settings[DatabaseSettingsName.encoding.value]
if DatabaseSettingsName.case_sensitive.value in settings:
self._case_sensitive = bool(settings[DatabaseSettingsName.case_sensitive.value])
if DatabaseSettingsName.echo.value in settings:
self._echo = bool(settings[DatabaseSettingsName.echo.value])
except Exception as e:
Console.set_foreground_color(ForegroundColor.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(ForegroundColor.default)

View File

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

View File

View File

@@ -0,0 +1,18 @@
from typing import Type
from sh_edraft.service.base.service_base import ServiceBase
class ProvideState:
def __init__(self, service: Type[ServiceBase] = None, args: tuple = None):
self._service: Type[ServiceBase] = service
self._args: tuple = args
@property
def service(self):
return self._service
@property
def args(self) -> tuple:
return self._args

View File

@@ -0,0 +1,7 @@
from abc import ABC, abstractmethod
class ServiceABC(ABC):
@abstractmethod
def __init__(self): pass

View File

@@ -0,0 +1,95 @@
from abc import ABC
from collections import Callable
from inspect import signature, Parameter
from typing import Type, Optional, Union
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.console.console import Console
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_abc import ServiceABC
from cpl.dependency_injection.service_provider_base import ServiceProviderABC
from cpl.environment.environment_abc import EnvironmentABC
class ServiceProvider(ServiceProviderABC):
def __init__(self, app_runtime: ApplicationRuntimeABC):
ServiceProviderABC.__init__(self)
self._app_runtime: ApplicationRuntimeABC = app_runtime
self._database_context: Optional[DatabaseContextABC] = None
self._transient_services: dict[Type[ServiceABC], Type[ServiceABC]] = {}
self._scoped_services: dict[Type[ServiceABC], Type[ServiceABC]] = {}
self._singleton_services: dict[Type[ServiceABC], Callable[ServiceABC], ServiceABC] = {}
def _create_instance(self, service: Union[Callable[ServiceABC], ServiceABC]) -> Callable[ServiceABC]:
sig = signature(service.__init__)
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if issubclass(parameter.annotation, ApplicationRuntimeABC):
params.append(self._app_runtime)
elif issubclass(parameter.annotation, EnvironmentABC):
params.append(self._app_runtime.configuration.environment)
elif issubclass(parameter.annotation, DatabaseContextABC):
params.append(self._database_context)
elif issubclass(parameter.annotation, ConfigurationModelABC):
params.append(self._app_runtime.configuration.get_configuration(parameter.annotation))
else:
params.append(self.get_service(parameter.annotation))
return service(*params)
def add_db_context(self, db_context: Type[DatabaseContextABC]):
self._database_context = self._create_instance(db_context)
def get_db_context(self) -> Callable[DatabaseContextABC]:
return self._database_context
def add_transient(self, service_type: Type[ServiceABC], service: Type[ServiceABC]):
self._transient_services[service_type] = service
def add_scoped(self, service_type: Type[ServiceABC], service: Type[ServiceABC]):
self._scoped_services[service_type] = service
def add_singleton(self, service_type: Type[ServiceABC], service: Callable[ServiceABC]):
for known_service in self._singleton_services:
if type(known_service) == service_type:
raise Exception(f'Service with type {service_type} already exists')
self._singleton_services[service_type] = self._create_instance(service)
def get_service(self, instance_type: Type) -> Callable[ServiceABC]:
for service in self._transient_services:
if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
return self._create_instance(self._transient_services[service])
for service in self._scoped_services:
if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
return self._create_instance(self._scoped_services[service])
for service in self._singleton_services:
if service == instance_type and isinstance(self._singleton_services[service], instance_type):
return self._singleton_services[service]
def remove_service(self, instance_type: Type[ServiceABC]):
for service in self._transient_services:
if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
del self._transient_services[service]
return
for service in self._scoped_services:
if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
del self._scoped_services[service]
return
for service in self._singleton_services:
if service == instance_type and isinstance(self._singleton_services[service], instance_type):
del self._singleton_services[service]
return

View File

@@ -0,0 +1,33 @@
from abc import abstractmethod, ABC
from collections import Callable
from typing import Type
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_abc import ServiceABC
class ServiceProviderABC(ABC):
@abstractmethod
def __init__(self): pass
@abstractmethod
def add_db_context(self, db_context: Type[DatabaseContextABC]): pass
@abstractmethod
def get_db_context(self) -> Callable[DatabaseContextABC]: pass
@abstractmethod
def add_transient(self, service_type: Type, service: Type): pass
@abstractmethod
def add_scoped(self, service_type: Type, service: Type): pass
@abstractmethod
def add_singleton(self, service_type: Type, service: Callable): pass
@abstractmethod
def get_service(self, instance_type: Type) -> Callable[ServiceABC]: pass
@abstractmethod
def remove_service(self, instance_type: type): pass

View File

View File

@@ -0,0 +1,43 @@
from abc import ABC, abstractmethod
class EnvironmentABC(ABC):
@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 content_root_path(self) -> str: pass
@content_root_path.setter
@abstractmethod
def content_root_path(self, content_root_path: str): pass
@property
@abstractmethod
def host_name(self) -> str: pass

View File

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

View File

@@ -0,0 +1,52 @@
from socket import gethostname
from typing import Optional
from cpl.environment.environment_abc import EnvironmentABC
from cpl.environment.environment_name import EnvironmentName
class HostingEnvironment(EnvironmentABC):
def __init__(self, name: EnvironmentName = EnvironmentName.production, crp: str = './'):
EnvironmentABC.__init__(self)
self._environment_name: Optional[EnvironmentName] = name
self._app_name: Optional[str] = None
self._customer: Optional[str] = None
self._content_root_path: Optional[str] = crp
@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 = EnvironmentName(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 content_root_path(self) -> str:
return self._content_root_path
@content_root_path.setter
def content_root_path(self, content_root_path: str):
self._content_root_path = content_root_path
@property
def host_name(self):
return gethostname()

View File

195
src/cpl/logging/logger.py Normal file
View File

@@ -0,0 +1,195 @@
import datetime
import os
import traceback
from string import Template
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.console.console import Console
from cpl.console.foreground_color import ForegroundColor
from cpl.logging.logger_abc import LoggerABC
from cpl.logging.logging_level import LoggingLevel
from cpl.logging.logging_settings import LoggingSettings
from cpl.time.time_format_settings import TimeFormatSettings
class Logger(LoggerABC):
def __init__(self, logging_settings: LoggingSettings, time_format: TimeFormatSettings, app_runtime: ApplicationRuntimeABC):
LoggerABC.__init__(self)
self._app_runtime = app_runtime
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._app_runtime.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_runtime.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:
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:
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:
""" 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}'
f = open(path, "w+")
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):
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: LoggingLevel, message: str) -> str:
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(ForegroundColor.default)
Console.write_line(string)
Console.set_foreground_color(ForegroundColor.default)
def trace(self, name: str, message: str):
output = self._get_string(name, LoggingLevel.TRACE, message)
# check if message can be written to log
if self._level.value >= LoggingLevel.TRACE.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevel.TRACE.value:
Console.set_foreground_color(ForegroundColor.green)
Console.write_line(output)
Console.set_foreground_color(ForegroundColor.default)
def debug(self, name: str, message: str):
output = self._get_string(name, LoggingLevel.DEBUG, message)
# check if message can be written to log
if self._level.value >= LoggingLevel.DEBUG.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevel.DEBUG.value:
Console.set_foreground_color(ForegroundColor.green)
Console.write_line(output)
Console.set_foreground_color(ForegroundColor.default)
def info(self, name: str, message: str):
output = self._get_string(name, LoggingLevel.INFO, message)
# check if message can be written to log
if self._level.value >= LoggingLevel.INFO.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevel.INFO.value:
Console.set_foreground_color(ForegroundColor.green)
Console.write_line(output)
Console.set_foreground_color(ForegroundColor.default)
def warn(self, name: str, message: str):
output = self._get_string(name, LoggingLevel.WARN, message)
# check if message can be written to log
if self._level.value >= LoggingLevel.WARN.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevel.WARN.value:
Console.set_foreground_color(ForegroundColor.yellow)
Console.write_line(output)
Console.set_foreground_color(ForegroundColor.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, LoggingLevel.ERROR, f'{ex} -> {tb}')
else:
output = self._get_string(name, LoggingLevel.ERROR, message)
# check if message can be written to log
if self._level.value >= LoggingLevel.ERROR.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevel.ERROR.value:
Console.set_foreground_color(ForegroundColor.red)
Console.write_line(output)
Console.set_foreground_color(ForegroundColor.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, LoggingLevel.FATAL, f'{ex} -> {tb}')
else:
output = self._get_string(name, LoggingLevel.FATAL, message)
# check if message can be written to log
if self._level.value >= LoggingLevel.FATAL.value:
self._append_log(output)
# check if message can be shown in console
if self._console.value >= LoggingLevel.FATAL.value:
Console.set_foreground_color(ForegroundColor.red)
Console.write_line(output)
Console.set_foreground_color(ForegroundColor.default)
exit()
def _fatal_console(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, LoggingLevel.ERROR, f'{ex} -> {tb}')
else:
output = self._get_string(name, LoggingLevel.ERROR, message)
# check if message can be shown in console
if self._console.value >= LoggingLevel.FATAL.value:
Console.set_foreground_color(ForegroundColor.red)
Console.write_line(output)
Console.set_foreground_color(ForegroundColor.default)
exit()

View File

@@ -0,0 +1,31 @@
from abc import abstractmethod
from cpl.dependency_injection.service_abc import ServiceABC
class LoggerABC(ServiceABC):
@abstractmethod
def __init__(self):
ServiceABC.__init__(self)
@abstractmethod
def header(self, string: str): pass
@abstractmethod
def trace(self, name: str, message: str): pass
@abstractmethod
def debug(self, name: str, message: str): pass
@abstractmethod
def info(self, name: str, message: str): pass
@abstractmethod
def warn(self, name: str, message: str): pass
@abstractmethod
def error(self, name: str, message: str, ex: Exception = None): pass
@abstractmethod
def fatal(self, name: str, message: str, ex: Exception = None): pass

View File

@@ -0,0 +1,12 @@
from enum import Enum
class LoggingLevel(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

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 import ForegroundColor
from cpl.logging.logging_level import LoggingLevel
from cpl.logging.logging_settings_name import LoggingSettingsName
class LoggingSettings(ConfigurationModelABC):
def __init__(self):
ConfigurationModelABC.__init__(self)
self._path: Optional[str] = None
self._filename: Optional[str] = None
self._console: Optional[LoggingLevel] = None
self._level: Optional[LoggingLevel] = 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) -> LoggingLevel:
return self._console
@console.setter
def console(self, console: LoggingLevel) -> None:
self._console = console
@property
def level(self) -> LoggingLevel:
return self._level
@level.setter
def level(self, level: LoggingLevel) -> None:
self._level = level
def from_dict(self, settings: dict):
try:
self._path = settings[LoggingSettingsName.path.value]
self._filename = settings[LoggingSettingsName.filename.value]
self._console = LoggingLevel[settings[LoggingSettingsName.console_level.value]]
self._level = LoggingLevel[settings[LoggingSettingsName.file_level.value]]
except Exception as e:
Console.set_foreground_color(ForegroundColor.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(ForegroundColor.default)

View File

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

View File

86
src/cpl/mailing/email.py Normal file
View File

@@ -0,0 +1,86 @@
import re
class EMail:
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:
return bool(re.search('^\\w+([.-]?\\w+)*@\\w+([.-]?\\w+)*(.\\w{2,3})+$', address))
def add_header(self, header: str):
if self._header is None:
self._header = []
self._header.append(header)
def add_receiver(self, receiver: str):
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):
return str(f'From: {transceiver}\r\nTo: {self.receiver}\r\n{self.header}\r\nSubject: {self.subject}\r\n{self.body}').encode('utf-8')

View File

@@ -0,0 +1,70 @@
import ssl
from smtplib import SMTP
from typing import Optional
from cpl.environment.environment_abc import EnvironmentABC
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):
def __init__(self, environment: EnvironmentABC, 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):
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):
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')

View File

@@ -0,0 +1,17 @@
from abc import abstractmethod
from cpl.dependency_injection.service_abc import ServiceABC
from cpl.mailing.email import EMail
class EMailClientABC(ServiceABC):
@abstractmethod
def __init__(self):
ServiceABC.__init__(self)
@abstractmethod
def connect(self): pass
@abstractmethod
def send_mail(self, email: EMail): pass

View File

@@ -0,0 +1,59 @@
import traceback
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.console.console import Console
from cpl.mailing.email_client_settings_name import EMailClientSettingsName
class EMailClientSettings(ConfigurationModelABC):
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[EMailClientSettingsName.host.value]
self._port = settings[EMailClientSettingsName.port.value]
self._user_name = settings[EMailClientSettingsName.user_name.value]
self._credentials = settings[EMailClientSettingsName.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()}')

View File

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

0
src/cpl/time/__init__.py Normal file
View File

View File

@@ -0,0 +1,61 @@
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 import ForegroundColor
from cpl.time.time_format_settings_names import TimeFormatSettingsNames
class TimeFormatSettings(ConfigurationModelABC):
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[TimeFormatSettingsNames.date_format.value]
self._time_format = settings[TimeFormatSettingsNames.time_format.value]
self._date_time_format = settings[TimeFormatSettingsNames.date_time_format.value]
self._date_time_log_format = settings[TimeFormatSettingsNames.date_time_log_format.value]
except Exception as e:
Console.set_foreground_color(ForegroundColor.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(ForegroundColor.default)

View File

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

View File

View File

@@ -0,0 +1,17 @@
import base64
class CredentialManager:
@staticmethod
def encrypt(string: str) -> str:
return base64.b64encode(string.encode('utf-8')).decode('utf-8')
@staticmethod
def decrypt(string: str) -> str:
return base64.b64decode(string).decode('utf-8')
@staticmethod
def build_string(string: str, credentials: str):
return string.replace('$credentials', CredentialManager.decrypt(credentials))