2023.4 #168
@ -25,6 +25,7 @@ from cpl_core.environment.application_environment import ApplicationEnvironment
|
||||
from cpl_core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
from cpl_core.environment.environment_name_enum import EnvironmentNameEnum
|
||||
from cpl_core.type import T
|
||||
from cpl_core.utils.json_processor import JSONProcessor
|
||||
|
||||
|
||||
class Configuration(ConfigurationABC):
|
||||
@ -279,7 +280,18 @@ class Configuration(ConfigurationABC):
|
||||
for key, value in config_from_file.items():
|
||||
if sub.__name__ == key or sub.__name__.replace("Settings", "") == key:
|
||||
configuration = sub()
|
||||
from_dict = getattr(configuration, "from_dict", None)
|
||||
|
||||
if from_dict is not None and not hasattr(from_dict, "is_base_func"):
|
||||
Console.set_foreground_color(ForegroundColorEnum.yellow)
|
||||
Console.write_line(
|
||||
f"{sub.__name__}.from_dict is deprecated. Instead, set attributes as typed arguments in __init__. They can be None by default!"
|
||||
)
|
||||
Console.color_reset()
|
||||
configuration.from_dict(value)
|
||||
else:
|
||||
configuration = JSONProcessor.process(sub, value)
|
||||
|
||||
self.add_configuration(sub, configuration)
|
||||
|
||||
def add_configuration(self, key_type: T, value: any):
|
||||
|
@ -1,13 +1,18 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
def base_func(method):
|
||||
method.is_base_func = True
|
||||
return method
|
||||
|
||||
|
||||
class ConfigurationModelABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
r"""ABC for settings representation"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
@base_func
|
||||
def from_dict(self, settings: dict):
|
||||
r"""Converts attributes to dict
|
||||
|
||||
|
@ -1,27 +1,34 @@
|
||||
import traceback
|
||||
from typing import Optional
|
||||
|
||||
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
|
||||
from cpl_core.console.console import Console
|
||||
from cpl_core.console.foreground_color_enum import ForegroundColorEnum
|
||||
from cpl_core.database.database_settings_name_enum import DatabaseSettingsNameEnum
|
||||
|
||||
|
||||
class DatabaseSettings(ConfigurationModelABC):
|
||||
r"""Represents settings for the database connection"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(
|
||||
self,
|
||||
host: str = None,
|
||||
port: int = None,
|
||||
user: str = None,
|
||||
password: str = None,
|
||||
databse: str = None,
|
||||
charset: str = None,
|
||||
use_unicode: bool = None,
|
||||
buffered: bool = None,
|
||||
auth_plugin: bool = None,
|
||||
):
|
||||
ConfigurationModelABC.__init__(self)
|
||||
|
||||
self._host: Optional[str] = None
|
||||
self._port: Optional[int] = None
|
||||
self._user: Optional[str] = None
|
||||
self._password: Optional[str] = None
|
||||
self._databse: Optional[str] = None
|
||||
self._charset: Optional[str] = None
|
||||
self._use_unicode: Optional[bool] = None
|
||||
self._buffered: Optional[bool] = None
|
||||
self._auth_plugin: Optional[str] = None
|
||||
self._host: Optional[str] = host
|
||||
self._port: Optional[int] = port
|
||||
self._user: Optional[str] = user
|
||||
self._password: Optional[str] = password
|
||||
self._databse: Optional[str] = databse
|
||||
self._charset: Optional[str] = charset
|
||||
self._use_unicode: Optional[bool] = use_unicode
|
||||
self._buffered: Optional[bool] = buffered
|
||||
self._auth_plugin: Optional[str] = auth_plugin
|
||||
|
||||
@property
|
||||
def host(self) -> Optional[str]:
|
||||
@ -59,35 +66,35 @@ class DatabaseSettings(ConfigurationModelABC):
|
||||
def auth_plugin(self) -> Optional[str]:
|
||||
return self._auth_plugin
|
||||
|
||||
def from_dict(self, settings: dict):
|
||||
r"""Sets attributes from given dict
|
||||
|
||||
Parameter:
|
||||
settings: :class:`dict`
|
||||
"""
|
||||
try:
|
||||
self._host = settings[DatabaseSettingsNameEnum.host.value]
|
||||
if DatabaseSettingsNameEnum.port.value in settings:
|
||||
self._port = settings[DatabaseSettingsNameEnum.port.value]
|
||||
else:
|
||||
self._port = 3306
|
||||
self._user = settings[DatabaseSettingsNameEnum.user.value]
|
||||
self._password = settings[DatabaseSettingsNameEnum.password.value]
|
||||
self._databse = settings[DatabaseSettingsNameEnum.database.value]
|
||||
|
||||
if DatabaseSettingsNameEnum.charset.value in settings:
|
||||
self._charset = settings[DatabaseSettingsNameEnum.charset.value]
|
||||
|
||||
if DatabaseSettingsNameEnum.buffered.value in settings:
|
||||
self._use_unicode = bool(settings[DatabaseSettingsNameEnum.use_unicode.value])
|
||||
|
||||
if DatabaseSettingsNameEnum.buffered.value in settings:
|
||||
self._buffered = bool(settings[DatabaseSettingsNameEnum.buffered.value])
|
||||
|
||||
if DatabaseSettingsNameEnum.auth_plugin.value in settings:
|
||||
self._auth_plugin = settings[DatabaseSettingsNameEnum.auth_plugin.value]
|
||||
except Exception as e:
|
||||
Console.set_foreground_color(ForegroundColorEnum.red)
|
||||
Console.write_line(f"[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings")
|
||||
Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
# def from_dict(self, settings: dict):
|
||||
# r"""Sets attributes from given dict
|
||||
#
|
||||
# Parameter:
|
||||
# settings: :class:`dict`
|
||||
# """
|
||||
# try:
|
||||
# self._host = settings[DatabaseSettingsNameEnum.host.value]
|
||||
# if DatabaseSettingsNameEnum.port.value in settings:
|
||||
# self._port = settings[DatabaseSettingsNameEnum.port.value]
|
||||
# else:
|
||||
# self._port = 3306
|
||||
# self._user = settings[DatabaseSettingsNameEnum.user.value]
|
||||
# self._password = settings[DatabaseSettingsNameEnum.password.value]
|
||||
# self._databse = settings[DatabaseSettingsNameEnum.database.value]
|
||||
#
|
||||
# if DatabaseSettingsNameEnum.charset.value in settings:
|
||||
# self._charset = settings[DatabaseSettingsNameEnum.charset.value]
|
||||
#
|
||||
# if DatabaseSettingsNameEnum.buffered.value in settings:
|
||||
# self._use_unicode = bool(settings[DatabaseSettingsNameEnum.use_unicode.value])
|
||||
#
|
||||
# if DatabaseSettingsNameEnum.buffered.value in settings:
|
||||
# self._buffered = bool(settings[DatabaseSettingsNameEnum.buffered.value])
|
||||
#
|
||||
# if DatabaseSettingsNameEnum.auth_plugin.value in settings:
|
||||
# self._auth_plugin = settings[DatabaseSettingsNameEnum.auth_plugin.value]
|
||||
# except Exception as e:
|
||||
# Console.set_foreground_color(ForegroundColorEnum.red)
|
||||
# Console.write_line(f"[ ERROR ] [ {__name__} ]: Reading error in {type(self).__name__} settings")
|
||||
# Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
# Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
|
@ -11,12 +11,18 @@ from cpl_core.logging.logging_settings_name_enum import LoggingSettingsNameEnum
|
||||
class LoggingSettings(ConfigurationModelABC):
|
||||
r"""Representation of logging settings"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(
|
||||
self,
|
||||
path: str = None,
|
||||
filename: str = None,
|
||||
console_log_level: LoggingLevelEnum = None,
|
||||
file_log_level: LoggingLevelEnum = None,
|
||||
):
|
||||
ConfigurationModelABC.__init__(self)
|
||||
self._path: Optional[str] = None
|
||||
self._filename: Optional[str] = None
|
||||
self._console: Optional[LoggingLevelEnum] = None
|
||||
self._level: Optional[LoggingLevelEnum] = None
|
||||
self._path: Optional[str] = path
|
||||
self._filename: Optional[str] = filename
|
||||
self._console: Optional[LoggingLevelEnum] = console_log_level
|
||||
self._level: Optional[LoggingLevelEnum] = file_log_level
|
||||
|
||||
@property
|
||||
def path(self) -> str:
|
||||
@ -50,14 +56,14 @@ class LoggingSettings(ConfigurationModelABC):
|
||||
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 {type(self).__name__} settings")
|
||||
Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
# 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 {type(self).__name__} settings")
|
||||
# Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
# Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
|
@ -1,20 +1,22 @@
|
||||
import traceback
|
||||
|
||||
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
|
||||
from cpl_core.console.console import Console
|
||||
from cpl_core.mailing.email_client_settings_name_enum import EMailClientSettingsNameEnum
|
||||
|
||||
|
||||
class EMailClientSettings(ConfigurationModelABC):
|
||||
r"""Representation of mailing settings"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(
|
||||
self,
|
||||
host: str = None,
|
||||
port: int = None,
|
||||
user_name: str = None,
|
||||
credentials: str = None,
|
||||
):
|
||||
ConfigurationModelABC.__init__(self)
|
||||
|
||||
self._host: str = ""
|
||||
self._port: int = 0
|
||||
self._user_name: str = ""
|
||||
self._credentials: str = ""
|
||||
self._host: str = host
|
||||
self._port: int = port
|
||||
self._user_name: str = user_name
|
||||
self._credentials: str = credentials
|
||||
|
||||
@property
|
||||
def host(self) -> str:
|
||||
@ -48,12 +50,12 @@ class EMailClientSettings(ConfigurationModelABC):
|
||||
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 {type(self).__name__} settings")
|
||||
Console.error(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
# 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 {type(self).__name__} settings")
|
||||
# Console.error(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
|
@ -10,12 +10,18 @@ from cpl_core.time.time_format_settings_names_enum import TimeFormatSettingsName
|
||||
class TimeFormatSettings(ConfigurationModelABC):
|
||||
r"""Representation of time format settings"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(
|
||||
self,
|
||||
date_format: str = None,
|
||||
time_format: str = None,
|
||||
date_time_format: str = None,
|
||||
date_time_log_format: str = None,
|
||||
):
|
||||
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
|
||||
self._date_format: Optional[str] = date_format
|
||||
self._time_format: Optional[str] = time_format
|
||||
self._date_time_format: Optional[str] = date_time_format
|
||||
self._date_time_log_format: Optional[str] = date_time_log_format
|
||||
|
||||
@property
|
||||
def date_format(self) -> str:
|
||||
@ -49,14 +55,14 @@ class TimeFormatSettings(ConfigurationModelABC):
|
||||
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 {type(self).__name__} settings")
|
||||
Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
# 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 {type(self).__name__} settings")
|
||||
# Console.write_line(f"[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}")
|
||||
# Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
|
41
src/cpl_core/utils/json_processor.py
Normal file
41
src/cpl_core/utils/json_processor.py
Normal file
@ -0,0 +1,41 @@
|
||||
import enum
|
||||
from inspect import signature, Parameter
|
||||
|
||||
from cpl_core.utils import String
|
||||
|
||||
|
||||
class JSONProcessor:
|
||||
@staticmethod
|
||||
def process(_t: type, values: dict) -> object:
|
||||
args = []
|
||||
|
||||
sig = signature(_t.__init__)
|
||||
for param in sig.parameters.items():
|
||||
parameter = param[1]
|
||||
if parameter.name == "self" or parameter.annotation == Parameter.empty:
|
||||
continue
|
||||
|
||||
name = String.first_to_upper(String.convert_to_camel_case(parameter.name))
|
||||
name_first_lower = String.first_to_lower(name)
|
||||
if name in values or name_first_lower in values:
|
||||
value = ""
|
||||
if name in values:
|
||||
value = values[name]
|
||||
else:
|
||||
value = values[name_first_lower]
|
||||
|
||||
if isinstance(value, dict):
|
||||
value = JSONProcessor.process(parameter.annotation, value)
|
||||
|
||||
if issubclass(parameter.annotation, enum.Enum):
|
||||
value = parameter.annotation[value]
|
||||
|
||||
args.append(value)
|
||||
|
||||
elif parameter.default != Parameter.empty:
|
||||
args.append(parameter.default)
|
||||
|
||||
else:
|
||||
args.append(None)
|
||||
|
||||
return _t(*args)
|
@ -57,4 +57,4 @@ class Application(ApplicationABC):
|
||||
Console.write_line("scope", scope)
|
||||
with self._services.create_scope() as s:
|
||||
Console.write_line("with scope", s)
|
||||
self.test_send_mail()
|
||||
# self.test_send_mail()
|
||||
|
Loading…
Reference in New Issue
Block a user