Compare commits
5 Commits
2025.09.15
...
2025.09.16
| Author | SHA1 | Date | |
|---|---|---|---|
| 58dbd3ed1e | |||
| cd7dfaf2b4 | |||
| 8ad3e3bdb4 | |||
| b97bc0a3ed | |||
| 5f25400bcd |
@@ -1,10 +1,9 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional
|
||||
|
||||
from cpl.core.configuration.configuration_abc import ConfigurationABC
|
||||
from cpl.dependency.service_provider_abc import ServiceProviderABC
|
||||
|
||||
from cpl.core.console.console import Console
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
|
||||
|
||||
class ApplicationABC(ABC):
|
||||
@@ -13,14 +12,12 @@ class ApplicationABC(ABC):
|
||||
Parameters:
|
||||
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
|
||||
Contains object loaded from appsettings
|
||||
services: :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
|
||||
services: :class:`cpl.dependency.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
|
||||
def __init__(self, services: ServiceProviderABC):
|
||||
self._services: Optional[ServiceProviderABC] = services
|
||||
|
||||
def run(self):
|
||||
@@ -49,12 +46,12 @@ class ApplicationABC(ABC):
|
||||
def configure(self):
|
||||
r"""Configure the application
|
||||
|
||||
Called by :class:`cpl.core.application.application_abc.ApplicationABC.run`
|
||||
Called by :class:`cpl.application.application_abc.ApplicationABC.run`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def main(self):
|
||||
r"""Custom entry point
|
||||
|
||||
Called by :class:`cpl.core.application.application_abc.ApplicationABC.run`
|
||||
Called by :class:`cpl.application.application_abc.ApplicationABC.run`
|
||||
"""
|
||||
97
src/cpl-application/cpl/application/application_builder.py
Normal file
97
src/cpl-application/cpl/application/application_builder.py
Normal file
@@ -0,0 +1,97 @@
|
||||
from typing import Type, Optional, Callable, Union
|
||||
|
||||
from cpl.application.application_abc import ApplicationABC
|
||||
from cpl.application.application_builder_abc import ApplicationBuilderABC
|
||||
from cpl.application.application_extension_abc import ApplicationExtensionABC
|
||||
from cpl.application.async_application_extension_abc import AsyncApplicationExtensionABC
|
||||
from cpl.application.async_startup_abc import AsyncStartupABC
|
||||
from cpl.application.async_startup_extension_abc import AsyncStartupExtensionABC
|
||||
from cpl.application.startup_abc import StartupABC
|
||||
from cpl.application.startup_extension_abc import StartupExtensionABC
|
||||
from cpl.core.configuration.configuration import Configuration
|
||||
from cpl.dependency.service_collection import ServiceCollection
|
||||
from cpl.core.environment import Environment
|
||||
|
||||
|
||||
class ApplicationBuilder(ApplicationBuilderABC):
|
||||
r"""This is class is used to build an 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 | AsyncStartupABC] = None
|
||||
|
||||
self._services = ServiceCollection()
|
||||
|
||||
self._app_extensions: list[Type[ApplicationExtensionABC | AsyncApplicationExtensionABC]] = []
|
||||
self._startup_extensions: list[Type[StartupExtensionABC | AsyncStartupABC]] = []
|
||||
|
||||
def use_startup(self, startup: Type[StartupABC | AsyncStartupABC]) -> "ApplicationBuilder":
|
||||
self._startup = startup()
|
||||
return self
|
||||
|
||||
def use_extension(
|
||||
self,
|
||||
extension: Type[
|
||||
ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC
|
||||
],
|
||||
) -> "ApplicationBuilder":
|
||||
if (
|
||||
issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)
|
||||
) and extension not in self._app_extensions:
|
||||
self._app_extensions.append(extension)
|
||||
elif (
|
||||
issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)
|
||||
) and extension not in self._startup_extensions:
|
||||
self._startup_extensions.append(extension)
|
||||
|
||||
return self
|
||||
|
||||
def _build_startup(self):
|
||||
for ex in self._startup_extensions:
|
||||
extension = ex()
|
||||
extension.configure_configuration(Configuration, Environment)
|
||||
extension.configure_services(self._services, Environment)
|
||||
|
||||
if self._startup is not None:
|
||||
self._startup.configure_configuration(Configuration, Environment)
|
||||
self._startup.configure_services(self._services, Environment)
|
||||
|
||||
async def _build_async_startup(self):
|
||||
for ex in self._startup_extensions:
|
||||
extension = ex()
|
||||
await extension.configure_configuration(Configuration, Environment)
|
||||
await extension.configure_services(self._services, Environment)
|
||||
|
||||
if self._startup is not None:
|
||||
await self._startup.configure_configuration(Configuration, Environment)
|
||||
await self._startup.configure_services(self._services, Environment)
|
||||
|
||||
def build(self) -> ApplicationABC:
|
||||
self._build_startup()
|
||||
|
||||
config = Configuration
|
||||
services = self._services.build_service_provider()
|
||||
|
||||
for ex in self._app_extensions:
|
||||
extension = ex()
|
||||
extension.run(config, services)
|
||||
|
||||
return self._app(services)
|
||||
|
||||
async def build_async(self) -> ApplicationABC:
|
||||
await self._build_async_startup()
|
||||
|
||||
config = Configuration
|
||||
services = self._services.build_service_provider()
|
||||
|
||||
for ex in self._app_extensions:
|
||||
extension = ex()
|
||||
await extension.run(config, services)
|
||||
|
||||
return self._app(services)
|
||||
@@ -1,12 +1,12 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Type
|
||||
|
||||
from cpl.core.application.application_abc import ApplicationABC
|
||||
from cpl.core.application.startup_abc import StartupABC
|
||||
from cpl.application.application_abc import ApplicationABC
|
||||
from cpl.application.startup_abc import StartupABC
|
||||
|
||||
|
||||
class ApplicationBuilderABC(ABC):
|
||||
r"""ABC for the :class:`cpl.core.application.application_builder.ApplicationBuilder`"""
|
||||
r"""ABC for the :class:`cpl.application.application_builder.ApplicationBuilder`"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self, *args):
|
||||
@@ -17,7 +17,7 @@ class ApplicationBuilderABC(ABC):
|
||||
r"""Sets the custom startup class to use
|
||||
|
||||
Parameter:
|
||||
startup: Type[:class:`cpl.core.application.startup_abc.StartupABC`]
|
||||
startup: Type[:class:`cpl.application.startup_abc.StartupABC`]
|
||||
Startup class to use
|
||||
"""
|
||||
|
||||
@@ -26,7 +26,7 @@ class ApplicationBuilderABC(ABC):
|
||||
r"""Sets the custom startup class to use async
|
||||
|
||||
Parameter:
|
||||
startup: Type[:class:`cpl.core.application.startup_abc.StartupABC`]
|
||||
startup: Type[:class:`cpl.application.startup_abc.StartupABC`]
|
||||
Startup class to use
|
||||
"""
|
||||
|
||||
@@ -35,7 +35,7 @@ class ApplicationBuilderABC(ABC):
|
||||
r"""Creates custom application object
|
||||
|
||||
Returns:
|
||||
Object of :class:`cpl.core.application.application_abc.ApplicationABC`
|
||||
Object of :class:`cpl.application.application_abc.ApplicationABC`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
@@ -43,5 +43,5 @@ class ApplicationBuilderABC(ABC):
|
||||
r"""Creates custom application object async
|
||||
|
||||
Returns:
|
||||
Object of :class:`cpl.core.application.application_abc.ApplicationABC`
|
||||
Object of :class:`cpl.application.application_abc.ApplicationABC`
|
||||
"""
|
||||
@@ -0,0 +1,14 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration.configuration import Configuration
|
||||
from cpl.dependency import ServiceProviderABC
|
||||
|
||||
|
||||
class ApplicationExtensionABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def run(self, config: Configuration, services: ServiceProviderABC):
|
||||
pass
|
||||
@@ -0,0 +1,14 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration.configuration import Configuration
|
||||
from cpl.dependency import ServiceProviderABC
|
||||
|
||||
|
||||
class AsyncApplicationExtensionABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def run(self, config: Configuration, services: ServiceProviderABC):
|
||||
pass
|
||||
23
src/cpl-application/cpl/application/async_startup_abc.py
Normal file
23
src/cpl-application/cpl/application/async_startup_abc.py
Normal file
@@ -0,0 +1,23 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.dependency.service_collection import ServiceCollection
|
||||
|
||||
|
||||
class AsyncStartupABC(ABC):
|
||||
r"""ABC for the startup class"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def configure_configuration(self):
|
||||
r"""Creates configuration of application"""
|
||||
|
||||
@abstractmethod
|
||||
async def configure_services(self, service: ServiceCollection):
|
||||
r"""Creates service provider
|
||||
|
||||
Parameter:
|
||||
services: :class:`cpl.dependency.service_collection`
|
||||
"""
|
||||
@@ -1,8 +1,8 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration.configuration_abc import ConfigurationABC
|
||||
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
from cpl.core.configuration.configuration import Configuration
|
||||
from cpl.dependency.service_collection import ServiceCollection
|
||||
from cpl.core.environment.environment import Environment
|
||||
|
||||
|
||||
class AsyncStartupExtensionABC(ABC):
|
||||
@@ -13,19 +13,19 @@ class AsyncStartupExtensionABC(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
|
||||
async def configure_configuration(self, config: Configuration, env: Environment):
|
||||
r"""Creates configuration of application
|
||||
|
||||
Parameter:
|
||||
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
|
||||
config: :class:`cpl.core.configuration.configuration_abc.Configuration`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
async def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
|
||||
async def configure_services(self, service: ServiceCollection, env: Environment):
|
||||
r"""Creates service provider
|
||||
|
||||
Parameter:
|
||||
services: :class:`cpl.core.dependency_injection.service_collection_abc`
|
||||
services: :class:`cpl.dependency.service_collection`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
31
src/cpl-application/cpl/application/startup_abc.py
Normal file
31
src/cpl-application/cpl/application/startup_abc.py
Normal file
@@ -0,0 +1,31 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration import Configuration
|
||||
from cpl.dependency.service_collection import ServiceCollection
|
||||
from cpl.core.environment import Environment
|
||||
|
||||
|
||||
class StartupABC(ABC):
|
||||
r"""ABC for the startup class"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def configure_configuration(self, config: Configuration, env: Environment):
|
||||
r"""Creates configuration of application
|
||||
|
||||
Parameter:
|
||||
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def configure_services(self, service: ServiceCollection, env: Environment):
|
||||
r"""Creates service provider
|
||||
|
||||
Parameter:
|
||||
services: :class:`cpl.dependency.service_collection`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
@@ -1,8 +1,10 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration.configuration_abc import ConfigurationABC
|
||||
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
|
||||
from cpl.core.configuration import Configuration
|
||||
from cpl.dependency.service_collection import ServiceCollection
|
||||
|
||||
from cpl.core.environment.environment import Environment
|
||||
|
||||
|
||||
class StartupExtensionABC(ABC):
|
||||
@@ -13,7 +15,7 @@ class StartupExtensionABC(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
|
||||
def configure_configuration(self, config: Configuration, env: Environment):
|
||||
r"""Creates configuration of application
|
||||
|
||||
Parameter:
|
||||
@@ -22,10 +24,10 @@ class StartupExtensionABC(ABC):
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
|
||||
def configure_services(self, service: ServiceCollection, env: Environment):
|
||||
r"""Creates service provider
|
||||
|
||||
Parameter:
|
||||
services: :class:`cpl.core.dependency_injection.service_collection_abc`
|
||||
services: :class:`cpl.dependency.service_collection`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
30
src/cpl-application/pyproject.toml
Normal file
30
src/cpl-application/pyproject.toml
Normal file
@@ -0,0 +1,30 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "cpl-application"
|
||||
version = "2024.7.0"
|
||||
description = "CPL application"
|
||||
readme ="CPL application package"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
authors = [
|
||||
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
|
||||
]
|
||||
keywords = ["cpl", "application", "backend", "shared", "library"]
|
||||
|
||||
dynamic = ["dependencies", "optional-dependencies"]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://www.sh-edraft.de"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
include = ["cpl*"]
|
||||
|
||||
[tool.setuptools.dynamic]
|
||||
dependencies = { file = ["requirements.txt"] }
|
||||
optional-dependencies.dev = { file = ["requirements.dev.txt"] }
|
||||
|
||||
|
||||
1
src/cpl-application/requirements.dev.txt
Normal file
1
src/cpl-application/requirements.dev.txt
Normal file
@@ -0,0 +1 @@
|
||||
black==25.1.0
|
||||
2
src/cpl-application/requirements.txt
Normal file
2
src/cpl-application/requirements.txt
Normal file
@@ -0,0 +1,2 @@
|
||||
cpl-core
|
||||
cpl-dependency
|
||||
@@ -1,91 +0,0 @@
|
||||
from typing import Type, Optional, Callable, Union
|
||||
|
||||
from cpl.core.application.application_abc import ApplicationABC
|
||||
from cpl.core.application.application_builder_abc import ApplicationBuilderABC
|
||||
from cpl.core.application.application_extension_abc import ApplicationExtensionABC
|
||||
from cpl.core.application.async_application_extension_abc import AsyncApplicationExtensionABC
|
||||
from cpl.core.application.async_startup_abc import AsyncStartupABC
|
||||
from cpl.core.application.async_startup_extension_abc import AsyncStartupExtensionABC
|
||||
from cpl.core.application.startup_abc import StartupABC
|
||||
from cpl.core.application.startup_extension_abc import StartupExtensionABC
|
||||
from cpl.core.configuration.configuration import Configuration
|
||||
from cpl.core.dependency_injection.service_collection import ServiceCollection
|
||||
|
||||
|
||||
class ApplicationBuilder(ApplicationBuilderABC):
|
||||
r"""This is class is used to build an object of :class:`cpl.core.application.application_abc.ApplicationABC`
|
||||
|
||||
Parameter:
|
||||
app: Type[:class:`cpl.core.application.application_abc.ApplicationABC`]
|
||||
Application to build
|
||||
"""
|
||||
|
||||
def __init__(self, app: Type[ApplicationABC]):
|
||||
ApplicationBuilderABC.__init__(self)
|
||||
self._app = app
|
||||
self._startup: Optional[StartupABC | AsyncStartupABC] = None
|
||||
|
||||
self._configuration = Configuration()
|
||||
self._environment = self._configuration.environment
|
||||
self._services = ServiceCollection(self._configuration)
|
||||
|
||||
self._app_extensions: list[Type[ApplicationExtensionABC | AsyncApplicationExtensionABC]] = []
|
||||
self._startup_extensions: list[Type[StartupExtensionABC | AsyncStartupABC]] = []
|
||||
|
||||
def use_startup(self, startup: Type[StartupABC | AsyncStartupABC]) -> "ApplicationBuilder":
|
||||
self._startup = startup()
|
||||
return self
|
||||
|
||||
def use_extension(
|
||||
self, extension: Type[ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC]
|
||||
) -> "ApplicationBuilder":
|
||||
if (issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)) and extension not in self._app_extensions:
|
||||
self._app_extensions.append(extension)
|
||||
elif (issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)) and extension not in self._startup_extensions:
|
||||
self._startup_extensions.append(extension)
|
||||
|
||||
return self
|
||||
|
||||
def _build_startup(self):
|
||||
for ex in self._startup_extensions:
|
||||
extension = ex()
|
||||
extension.configure_configuration(self._configuration, self._environment)
|
||||
extension.configure_services(self._services, self._environment)
|
||||
|
||||
if self._startup is not None:
|
||||
self._startup.configure_configuration(self._configuration, self._environment)
|
||||
self._startup.configure_services(self._services, self._environment)
|
||||
|
||||
async def _build_async_startup(self):
|
||||
for ex in self._startup_extensions:
|
||||
extension = ex()
|
||||
await extension.configure_configuration(self._configuration, self._environment)
|
||||
await extension.configure_services(self._services, self._environment)
|
||||
|
||||
if self._startup is not None:
|
||||
await self._startup.configure_configuration(self._configuration, self._environment)
|
||||
await self._startup.configure_services(self._services, self._environment)
|
||||
|
||||
def build(self) -> ApplicationABC:
|
||||
self._build_startup()
|
||||
|
||||
config = self._configuration
|
||||
services = self._services.build_service_provider()
|
||||
|
||||
for ex in self._app_extensions:
|
||||
extension = ex()
|
||||
extension.run(config, services)
|
||||
|
||||
return self._app(config, services)
|
||||
|
||||
async def build_async(self) -> ApplicationABC:
|
||||
await self._build_async_startup()
|
||||
|
||||
config = self._configuration
|
||||
services = self._services.build_service_provider()
|
||||
|
||||
for ex in self._app_extensions:
|
||||
extension = ex()
|
||||
await extension.run(config, services)
|
||||
|
||||
return self._app(config, services)
|
||||
@@ -1,14 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration import ConfigurationABC
|
||||
from cpl.core.dependency_injection import ServiceProviderABC
|
||||
|
||||
|
||||
class ApplicationExtensionABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def run(self, config: ConfigurationABC, services: ServiceProviderABC):
|
||||
pass
|
||||
@@ -1,14 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration import ConfigurationABC
|
||||
from cpl.core.dependency_injection import ServiceProviderABC
|
||||
|
||||
|
||||
class AsyncApplicationExtensionABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def run(self, config: ConfigurationABC, services: ServiceProviderABC):
|
||||
pass
|
||||
@@ -1,32 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration.configuration_abc import ConfigurationABC
|
||||
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
|
||||
|
||||
class AsyncStartupABC(ABC):
|
||||
r"""ABC for the startup class"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
|
||||
r"""Creates configuration of application
|
||||
|
||||
Parameter:
|
||||
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
async def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
|
||||
r"""Creates service provider
|
||||
|
||||
Parameter:
|
||||
services: :class:`cpl.core.dependency_injection.service_collection_abc`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
@@ -1,32 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration.configuration_abc import ConfigurationABC
|
||||
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
|
||||
|
||||
class StartupABC(ABC):
|
||||
r"""ABC for the startup class"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def configure_configuration(self, config: ConfigurationABC, env: ApplicationEnvironmentABC):
|
||||
r"""Creates configuration of application
|
||||
|
||||
Parameter:
|
||||
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def configure_services(self, service: ServiceCollectionABC, env: ApplicationEnvironmentABC):
|
||||
r"""Creates service provider
|
||||
|
||||
Parameter:
|
||||
services: :class:`cpl.core.dependency_injection.service_collection_abc`
|
||||
env: :class:`cpl.core.environment.application_environment_abc`
|
||||
"""
|
||||
@@ -1,12 +1,2 @@
|
||||
from .argument_abc import ArgumentABC
|
||||
from .argument_builder import ArgumentBuilder
|
||||
from .argument_executable_abc import ArgumentExecutableABC
|
||||
from .argument_type_enum import ArgumentTypeEnum
|
||||
from .configuration import Configuration
|
||||
from .configuration_abc import ConfigurationABC
|
||||
from .configuration_model_abc import ConfigurationModelABC
|
||||
from .configuration_variable_name_enum import ConfigurationVariableNameEnum
|
||||
from .executable_argument import ExecutableArgument
|
||||
from .flag_argument import FlagArgument
|
||||
from .validator_abc import ValidatorABC
|
||||
from .variable_argument import VariableArgument
|
||||
|
||||
@@ -1,64 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.configuration.argument_type_enum import ArgumentTypeEnum
|
||||
|
||||
|
||||
class ArgumentABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(
|
||||
self,
|
||||
token: str,
|
||||
name: str,
|
||||
aliases: list[str],
|
||||
prevent_next_executable: bool = False,
|
||||
console_arguments: list["ArgumentABC"] = None,
|
||||
):
|
||||
r"""Representation of an console argument
|
||||
|
||||
Parameter:
|
||||
token: :class:`str`
|
||||
name: :class:`str`
|
||||
aliases: list[:class:`str`]
|
||||
console_arguments: List[:class:`cpl.core.configuration.console_argument.ConsoleArgument`]
|
||||
"""
|
||||
self._token = token
|
||||
self._name = name
|
||||
self._aliases = aliases
|
||||
self._prevent_next_executable = prevent_next_executable
|
||||
self._console_arguments = console_arguments if console_arguments is not None else []
|
||||
|
||||
@property
|
||||
def token(self) -> str:
|
||||
return self._token
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def aliases(self) -> list[str]:
|
||||
return self._aliases
|
||||
|
||||
@property
|
||||
def prevent_next_executable(self) -> bool:
|
||||
return self._prevent_next_executable
|
||||
|
||||
@property
|
||||
def console_arguments(self) -> list["ArgumentABC"]:
|
||||
return self._console_arguments
|
||||
|
||||
def add_console_argument(self, arg_type: ArgumentTypeEnum, *args, **kwargs) -> "ArgumentABC":
|
||||
r"""Creates and adds a console argument to known console arguments
|
||||
|
||||
Parameter:
|
||||
arg_type: :class:`str`
|
||||
Specifies the specific type of the argument
|
||||
|
||||
Returns:
|
||||
self :class:`cpl.core.configuration.console_argument.ConsoleArgument` not created argument!
|
||||
"""
|
||||
from cpl.core.configuration.argument_builder import ArgumentBuilder
|
||||
|
||||
argument = ArgumentBuilder.build_argument(arg_type, *args, *kwargs)
|
||||
self._console_arguments.append(argument)
|
||||
return self
|
||||
@@ -1,30 +0,0 @@
|
||||
from typing import Union
|
||||
|
||||
from cpl.core.configuration.argument_type_enum import ArgumentTypeEnum
|
||||
from cpl.core.configuration.executable_argument import ExecutableArgument
|
||||
from cpl.core.configuration.flag_argument import FlagArgument
|
||||
from cpl.core.configuration.variable_argument import VariableArgument
|
||||
from cpl.core.console import Console
|
||||
|
||||
|
||||
class ArgumentBuilder:
|
||||
@staticmethod
|
||||
def build_argument(
|
||||
arg_type: ArgumentTypeEnum, *args, **kwargs
|
||||
) -> Union[ExecutableArgument, FlagArgument, VariableArgument]:
|
||||
argument = None
|
||||
try:
|
||||
match arg_type:
|
||||
case ArgumentTypeEnum.Flag:
|
||||
argument = FlagArgument(*args, **kwargs)
|
||||
case ArgumentTypeEnum.Executable:
|
||||
argument = ExecutableArgument(*args, **kwargs)
|
||||
case ArgumentTypeEnum.Variable:
|
||||
argument = VariableArgument(*args, **kwargs)
|
||||
case _:
|
||||
Console.error("Invalid argument type")
|
||||
Console.close()
|
||||
except TypeError as e:
|
||||
Console.error(str(e))
|
||||
Console.close()
|
||||
return argument
|
||||
@@ -1,11 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class ArgumentExecutableABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def run(self, args: list[str]):
|
||||
pass
|
||||
@@ -1,7 +0,0 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class ArgumentTypeEnum(Enum):
|
||||
Flag = 0
|
||||
Executable = 1
|
||||
Variable = 3
|
||||
@@ -1,65 +1,19 @@
|
||||
import inspect
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import traceback
|
||||
from collections.abc import Callable
|
||||
from typing import Union, Optional
|
||||
from typing import Any
|
||||
|
||||
from cpl.core.configuration.argument_abc import ArgumentABC
|
||||
from cpl.core.configuration.argument_builder import ArgumentBuilder
|
||||
from cpl.core.configuration.argument_type_enum import ArgumentTypeEnum
|
||||
from cpl.core.configuration.configuration_abc import ConfigurationABC
|
||||
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
|
||||
from cpl.core.configuration.configuration_variable_name_enum import (
|
||||
ConfigurationVariableNameEnum,
|
||||
)
|
||||
from cpl.core.configuration.executable_argument import ExecutableArgument
|
||||
from cpl.core.configuration.flag_argument import FlagArgument
|
||||
from cpl.core.configuration.variable_argument import VariableArgument
|
||||
from cpl.core.console.console import Console
|
||||
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
from cpl.core.environment.application_environment import ApplicationEnvironment
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
from cpl.core.environment.environment_name_enum import EnvironmentNameEnum
|
||||
from cpl.core.typing import T, R
|
||||
from cpl.core.environment.environment import Environment
|
||||
from cpl.core.typing import D, T
|
||||
from cpl.core.utils.json_processor import JSONProcessor
|
||||
|
||||
|
||||
class Configuration(ConfigurationABC):
|
||||
def __init__(self):
|
||||
r"""Representation of configuration"""
|
||||
ConfigurationABC.__init__(self)
|
||||
|
||||
self._application_environment = ApplicationEnvironment()
|
||||
self._config: dict[Union[type, str], Union[ConfigurationModelABC, str]] = {}
|
||||
|
||||
self._argument_types: list[ArgumentABC] = []
|
||||
self._additional_arguments: list[str] = []
|
||||
|
||||
self._argument_error_function: Optional[Callable] = None
|
||||
|
||||
self._handled_args = []
|
||||
|
||||
@property
|
||||
def environment(self) -> ApplicationEnvironmentABC:
|
||||
return self._application_environment
|
||||
|
||||
@property
|
||||
def additional_arguments(self) -> list[str]:
|
||||
return self._additional_arguments
|
||||
|
||||
@property
|
||||
def argument_error_function(self) -> Optional[Callable]:
|
||||
return self._argument_error_function
|
||||
|
||||
@argument_error_function.setter
|
||||
def argument_error_function(self, argument_error_function: Callable):
|
||||
self._argument_error_function = argument_error_function
|
||||
|
||||
@property
|
||||
def arguments(self) -> list[ArgumentABC]:
|
||||
return self._argument_types
|
||||
class Configuration:
|
||||
_config = {}
|
||||
|
||||
@staticmethod
|
||||
def _print_info(message: str):
|
||||
@@ -72,7 +26,7 @@ class Configuration(ConfigurationABC):
|
||||
Info message
|
||||
"""
|
||||
Console.set_foreground_color(ForegroundColorEnum.green)
|
||||
Console.write_line(f"[CONF] {message}")
|
||||
Console.write_line(f"[CONFIG] {message}")
|
||||
Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
|
||||
@staticmethod
|
||||
@@ -86,7 +40,7 @@ class Configuration(ConfigurationABC):
|
||||
Warning message
|
||||
"""
|
||||
Console.set_foreground_color(ForegroundColorEnum.yellow)
|
||||
Console.write_line(f"[CONF] {message}")
|
||||
Console.write_line(f"[CONFIG] {message}")
|
||||
Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
|
||||
@staticmethod
|
||||
@@ -100,31 +54,11 @@ class Configuration(ConfigurationABC):
|
||||
Error message
|
||||
"""
|
||||
Console.set_foreground_color(ForegroundColorEnum.red)
|
||||
Console.write_line(f"[CONF] {message}")
|
||||
Console.write_line(f"[CONFIG] {message}")
|
||||
Console.set_foreground_color(ForegroundColorEnum.default)
|
||||
|
||||
def _set_variable(self, name: str, value: any):
|
||||
r"""Sets variable to given value
|
||||
|
||||
Parameter:
|
||||
name: :class:`str`
|
||||
Name of the variable
|
||||
value: :class:`any`
|
||||
Value of the variable
|
||||
"""
|
||||
if name == ConfigurationVariableNameEnum.environment.value:
|
||||
self._application_environment.environment_name = EnvironmentNameEnum(value)
|
||||
|
||||
elif name == ConfigurationVariableNameEnum.name.value:
|
||||
self._application_environment.application_name = value
|
||||
|
||||
elif name == ConfigurationVariableNameEnum.customer.value:
|
||||
self._application_environment.customer = value
|
||||
|
||||
else:
|
||||
self._config[name] = value
|
||||
|
||||
def _load_json_file(self, file: str, output: bool) -> dict:
|
||||
@classmethod
|
||||
def _load_json_file(cls, file: str, output: bool) -> dict:
|
||||
r"""Reads the json file
|
||||
|
||||
Parameter:
|
||||
@@ -142,90 +76,19 @@ class Configuration(ConfigurationABC):
|
||||
# load json
|
||||
json_cfg = json.load(cfg)
|
||||
if output:
|
||||
self._print_info(f"Loaded config file: {file}")
|
||||
cls._print_info(f"Loaded config file: {file}")
|
||||
|
||||
return json_cfg
|
||||
except Exception as e:
|
||||
self._print_error(f"Cannot load config file: {file}! -> {e}")
|
||||
cls._print_error(f"Cannot load config file: {file}! -> {e}")
|
||||
return {}
|
||||
|
||||
def _parse_arguments(
|
||||
self,
|
||||
executables: list[ArgumentABC],
|
||||
arg_list: list[str],
|
||||
args_types: list[ArgumentABC],
|
||||
):
|
||||
for i in range(0, len(arg_list)):
|
||||
arg_str = arg_list[i]
|
||||
for n in range(0, len(args_types)):
|
||||
arg = args_types[n]
|
||||
arg_str_without_token = arg_str
|
||||
if arg.token != "" and arg.token in arg_str:
|
||||
arg_str_without_token = arg_str.split(arg.token)[1]
|
||||
|
||||
# executable
|
||||
if isinstance(arg, ExecutableArgument):
|
||||
if (
|
||||
arg_str.startswith(arg.token)
|
||||
and arg_str_without_token == arg.name
|
||||
or arg_str_without_token in arg.aliases
|
||||
):
|
||||
executables.append(arg)
|
||||
self._handled_args.append(arg_str)
|
||||
self._parse_arguments(executables, arg_list[i + 1 :], arg.console_arguments)
|
||||
|
||||
# variables
|
||||
elif isinstance(arg, VariableArgument):
|
||||
arg_str_without_value = arg_str_without_token
|
||||
if arg.value_token in arg_str_without_value:
|
||||
arg_str_without_value = arg_str_without_token.split(arg.value_token)[0]
|
||||
|
||||
if (
|
||||
arg_str.startswith(arg.token)
|
||||
and arg_str_without_value == arg.name
|
||||
or arg_str_without_value in arg.aliases
|
||||
):
|
||||
if arg.value_token != " ":
|
||||
value = arg_str_without_token.split(arg.value_token)[1]
|
||||
else:
|
||||
value = arg_list[i + 1]
|
||||
self._set_variable(arg.name, value)
|
||||
self._handled_args.append(arg_str)
|
||||
self._handled_args.append(value)
|
||||
self._parse_arguments(executables, arg_list[i + 1 :], arg.console_arguments)
|
||||
|
||||
# flags
|
||||
elif isinstance(arg, FlagArgument):
|
||||
if (
|
||||
arg_str.startswith(arg.token)
|
||||
and arg_str_without_token == arg.name
|
||||
or arg_str_without_token in arg.aliases
|
||||
):
|
||||
if arg_str in self._additional_arguments:
|
||||
self._additional_arguments.remove(arg_str)
|
||||
self._additional_arguments.append(arg.name)
|
||||
self._handled_args.append(arg_str)
|
||||
self._parse_arguments(executables, arg_list[i + 1 :], arg.console_arguments)
|
||||
|
||||
# add left over values to args
|
||||
if arg_str not in self._additional_arguments and arg_str not in self._handled_args:
|
||||
self._additional_arguments.append(arg_str)
|
||||
|
||||
def add_environment_variables(self, prefix: str):
|
||||
for env_var in os.environ.keys():
|
||||
if not env_var.startswith(prefix):
|
||||
continue
|
||||
|
||||
self._set_variable(env_var.replace(prefix, ""), os.environ[env_var])
|
||||
|
||||
def add_console_argument(self, argument: ArgumentABC):
|
||||
self._argument_types.append(argument)
|
||||
|
||||
def add_json_file(self, name: str, optional: bool = None, output: bool = True, path: str = None):
|
||||
@classmethod
|
||||
def add_json_file(cls, name: str, optional: bool = None, output: bool = True, path: str = None):
|
||||
if os.path.isabs(name):
|
||||
file_path = name
|
||||
else:
|
||||
path_root = self._application_environment.working_directory
|
||||
path_root = Environment.get_cwd()
|
||||
if path is not None:
|
||||
path_root = path
|
||||
|
||||
@@ -237,122 +100,35 @@ class Configuration(ConfigurationABC):
|
||||
if not os.path.isfile(file_path):
|
||||
if optional is not True:
|
||||
if output:
|
||||
self._print_error(f"File not found: {file_path}")
|
||||
cls._print_error(f"File not found: {file_path}")
|
||||
|
||||
sys.exit()
|
||||
|
||||
if output:
|
||||
self._print_warn(__name__, f"Not Loaded config file: {file_path}")
|
||||
cls._print_warn(f"Not Loaded config file: {file_path}")
|
||||
|
||||
return None
|
||||
|
||||
config_from_file = self._load_json_file(file_path, output)
|
||||
config_from_file = cls._load_json_file(file_path, output)
|
||||
for sub in ConfigurationModelABC.__subclasses__():
|
||||
for key, value in config_from_file.items():
|
||||
if sub.__name__ != key and sub.__name__.replace("Settings", "") != key:
|
||||
continue
|
||||
|
||||
configuration = sub()
|
||||
from_dict = getattr(configuration, "from_dict", None)
|
||||
configuration = JSONProcessor.process(sub, value)
|
||||
|
||||
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)
|
||||
cls.set(sub, configuration)
|
||||
|
||||
self.add_configuration(sub, configuration)
|
||||
@classmethod
|
||||
def set(cls, key: Any, value: T):
|
||||
if inspect.isclass(key):
|
||||
key = key.__name__
|
||||
|
||||
def add_configuration(self, key_type: T, value: any):
|
||||
self._config[key_type] = value
|
||||
cls._config[key] = value
|
||||
|
||||
def create_console_argument(
|
||||
self,
|
||||
arg_type: ArgumentTypeEnum,
|
||||
token: str,
|
||||
name: str,
|
||||
aliases: list[str],
|
||||
*args,
|
||||
**kwargs,
|
||||
) -> ArgumentABC:
|
||||
argument = ArgumentBuilder.build_argument(arg_type, token, name, aliases, *args, **kwargs)
|
||||
self._argument_types.append(argument)
|
||||
return argument
|
||||
@classmethod
|
||||
def get(cls, key: Any, default: D = None) -> T | D:
|
||||
if inspect.isclass(key):
|
||||
key = key.__name__
|
||||
|
||||
def for_each_argument(self, call: Callable):
|
||||
for arg in self._argument_types:
|
||||
call(arg)
|
||||
|
||||
def get_configuration(self, search_type: T) -> Optional[R]:
|
||||
if type(search_type) is str:
|
||||
if search_type == ConfigurationVariableNameEnum.environment.value:
|
||||
return self._application_environment.environment_name
|
||||
|
||||
elif search_type == ConfigurationVariableNameEnum.name.value:
|
||||
return self._application_environment.application_name
|
||||
|
||||
elif search_type == ConfigurationVariableNameEnum.customer.value:
|
||||
return self._application_environment.customer
|
||||
|
||||
if search_type not in self._config:
|
||||
return None
|
||||
|
||||
for config_model in self._config:
|
||||
if config_model == search_type:
|
||||
return self._config[config_model]
|
||||
|
||||
def parse_console_arguments(self, services: ServiceProviderABC, error: bool = None) -> bool:
|
||||
# sets environment variables as possible arguments as: --VAR=VALUE
|
||||
for arg_name in ConfigurationVariableNameEnum.to_list():
|
||||
self.add_console_argument(VariableArgument("--", str(arg_name).upper(), [str(arg_name).lower()], "="))
|
||||
|
||||
success = False
|
||||
try:
|
||||
arg_list = sys.argv[1:]
|
||||
executables: list[ExecutableArgument] = []
|
||||
self._parse_arguments(executables, arg_list, self._argument_types)
|
||||
except Exception:
|
||||
Console.error("An error occurred while parsing arguments.", traceback.format_exc())
|
||||
sys.exit()
|
||||
|
||||
try:
|
||||
prevent = False
|
||||
for exe in executables:
|
||||
if prevent:
|
||||
continue
|
||||
|
||||
if exe.validators is not None:
|
||||
abort = False
|
||||
for validator_type in exe.validators:
|
||||
validator = services.get_service(validator_type)
|
||||
result = validator.validate()
|
||||
abort = not result
|
||||
if abort:
|
||||
break
|
||||
|
||||
if abort:
|
||||
sys.exit()
|
||||
|
||||
cmd = services.get_service(exe.executable_type)
|
||||
self._handle_pre_or_post_executables(True, exe, services)
|
||||
self._set_variable("ACTIVE_EXECUTABLE", exe.name)
|
||||
args = self.get_configuration("ARGS")
|
||||
if args is not None:
|
||||
for arg in args.split(" "):
|
||||
if arg == "":
|
||||
continue
|
||||
self._additional_arguments.append(arg)
|
||||
|
||||
cmd.run(self._additional_arguments)
|
||||
self._handle_pre_or_post_executables(False, exe, services)
|
||||
prevent = exe.prevent_next_executable
|
||||
success = True
|
||||
except Exception:
|
||||
Console.error("An error occurred while executing arguments.", traceback.format_exc())
|
||||
sys.exit()
|
||||
|
||||
return success
|
||||
return cls._config.get(key, default)
|
||||
|
||||
@@ -1,140 +0,0 @@
|
||||
from abc import abstractmethod, ABC
|
||||
from collections.abc import Callable
|
||||
from typing import Optional
|
||||
|
||||
from cpl.core.configuration.argument_abc import ArgumentABC
|
||||
from cpl.core.configuration.argument_type_enum import ArgumentTypeEnum
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
from cpl.core.typing import T, R
|
||||
|
||||
|
||||
class ConfigurationABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
r"""ABC for the :class:`cpl.core.configuration.configuration.Configuration`"""
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def environment(self) -> ApplicationEnvironmentABC:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def additional_arguments(self) -> list[str]:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def argument_error_function(self) -> Optional[Callable]:
|
||||
pass
|
||||
|
||||
@argument_error_function.setter
|
||||
@abstractmethod
|
||||
def argument_error_function(self, argument_error_function: Callable):
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def arguments(self) -> list[ArgumentABC]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def add_environment_variables(self, prefix: str):
|
||||
r"""Reads the environment variables
|
||||
|
||||
Parameter:
|
||||
prefix: :class:`str`
|
||||
Prefix of the variables
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def add_console_argument(self, argument: ArgumentABC):
|
||||
r"""Adds console argument to known console arguments
|
||||
|
||||
Parameter:
|
||||
argument: :class:`cpl.core.configuration.console_argument.ConsoleArgumentABC`
|
||||
Specifies the console argument
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def add_json_file(self, name: str, optional: bool = None, output: bool = True, path: str = None):
|
||||
r"""Reads and saves settings from given json file
|
||||
|
||||
Parameter:
|
||||
name: :class:`str`
|
||||
Name of the file
|
||||
optional: :class:`str`
|
||||
Specifies whether an error should occur if the file was not found
|
||||
output: :class:`bool`
|
||||
Specifies whether an output should take place
|
||||
path: :class:`str`
|
||||
Path in which the file should be stored
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def add_configuration(self, key_type: T, value: any):
|
||||
r"""Add configuration object
|
||||
|
||||
Parameter:
|
||||
key_type: :class:`cpl.core.type.T`
|
||||
Type of the value
|
||||
value: any
|
||||
Object of the value
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def create_console_argument(
|
||||
self, arg_type: ArgumentTypeEnum, token: str, name: str, aliases: list[str], *args, **kwargs
|
||||
) -> ArgumentABC:
|
||||
r"""Creates and adds a console argument to known console arguments
|
||||
|
||||
Parameter:
|
||||
token: :class:`str`
|
||||
Specifies optional beginning of argument
|
||||
name :class:`str`
|
||||
Specifies name of argument
|
||||
aliases list[:class:`str`]
|
||||
Specifies possible aliases of name
|
||||
value_token :class:`str`
|
||||
Specifies were the value begins
|
||||
is_value_token_optional :class:`bool`
|
||||
Specifies if values are optional
|
||||
runnable: :class:`cpl.core.configuration.console_argument.ConsoleArgumentABC`
|
||||
Specifies class to run when called if value is not None
|
||||
|
||||
Returns:
|
||||
Object of :class:`cpl.core.configuration.console_argument.ConsoleArgumentABC`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def for_each_argument(self, call: Callable):
|
||||
r"""Iterates through all arguments and calls the call function
|
||||
|
||||
Parameter:
|
||||
call: :class:`Callable`
|
||||
Call for each argument
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_configuration(self, search_type: T) -> Optional[R]:
|
||||
r"""Returns value from configuration by given type
|
||||
|
||||
Parameter:
|
||||
search_type: :class:`cpl.core.type.T`
|
||||
Type to search for
|
||||
|
||||
Returns:
|
||||
Object of Union[:class:`str`, :class:`cpl.core.configuration.configuration_model_abc.ConfigurationModelABC`]
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def parse_console_arguments(self, services: "ServiceProviderABC", error: bool = None) -> bool:
|
||||
r"""Reads the console arguments
|
||||
|
||||
Parameter:
|
||||
error: :class:`bool`
|
||||
Defines is invalid argument error will be shown or not
|
||||
|
||||
Returns:
|
||||
Bool to specify if executables were executed or not.
|
||||
"""
|
||||
@@ -1,21 +1,5 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
def base_func(method):
|
||||
method.is_base_func = True
|
||||
return method
|
||||
from abc import ABC
|
||||
|
||||
|
||||
class ConfigurationModelABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
r"""ABC for settings representation"""
|
||||
|
||||
@base_func
|
||||
def from_dict(self, settings: dict):
|
||||
r"""DEPRECATED: Set attributes as typed arguments in __init__ instead. See https://docs.sh-edraft.de/cpl/deprecated.html#ConfigurationModelABC-from_dict-method for further information
|
||||
Converts attributes to dict
|
||||
|
||||
Parameter:
|
||||
settings: :class:`dict`
|
||||
"""
|
||||
pass
|
||||
|
||||
@@ -1,11 +0,0 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class ConfigurationVariableNameEnum(Enum):
|
||||
environment = "ENVIRONMENT"
|
||||
name = "NAME"
|
||||
customer = "CUSTOMER"
|
||||
|
||||
@staticmethod
|
||||
def to_list():
|
||||
return [var.value for var in ConfigurationVariableNameEnum]
|
||||
@@ -1,40 +0,0 @@
|
||||
from typing import Type, Optional
|
||||
|
||||
from cpl.core.configuration.argument_executable_abc import ArgumentExecutableABC
|
||||
from cpl.core.configuration.argument_abc import ArgumentABC
|
||||
from cpl.core.configuration.validator_abc import ValidatorABC
|
||||
|
||||
|
||||
class ExecutableArgument(ArgumentABC):
|
||||
def __init__(
|
||||
self,
|
||||
token: str,
|
||||
name: str,
|
||||
aliases: list[str],
|
||||
executable: Type[ArgumentExecutableABC],
|
||||
prevent_next_executable: bool = False,
|
||||
validators: list[Type[ValidatorABC]] = None,
|
||||
console_arguments: list["ArgumentABC"] = None,
|
||||
):
|
||||
self._executable_type = executable
|
||||
self._validators = validators
|
||||
self._executable: Optional[ArgumentExecutableABC] = None
|
||||
|
||||
ArgumentABC.__init__(self, token, name, aliases, prevent_next_executable, console_arguments)
|
||||
|
||||
@property
|
||||
def executable_type(self) -> type:
|
||||
return self._executable_type
|
||||
|
||||
def set_executable(self, executable: ArgumentExecutableABC):
|
||||
self._executable = executable
|
||||
|
||||
@property
|
||||
def validators(self) -> list[Type[ValidatorABC]]:
|
||||
return self._validators
|
||||
|
||||
def run(self, args: list[str]):
|
||||
r"""Executes runnable if exists"""
|
||||
if self._executable is None:
|
||||
return
|
||||
self._executable.execute(args)
|
||||
@@ -1,13 +0,0 @@
|
||||
from cpl.core.configuration.argument_abc import ArgumentABC
|
||||
|
||||
|
||||
class FlagArgument(ArgumentABC):
|
||||
def __init__(
|
||||
self,
|
||||
token: str,
|
||||
name: str,
|
||||
aliases: list[str],
|
||||
prevent_next_executable: bool = False,
|
||||
console_arguments: list["ArgumentABC"] = None,
|
||||
):
|
||||
ArgumentABC.__init__(self, token, name, aliases, prevent_next_executable, console_arguments)
|
||||
@@ -1,11 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class ValidatorABC(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def validate(self) -> bool:
|
||||
pass
|
||||
@@ -1,28 +0,0 @@
|
||||
from cpl.core.configuration.argument_abc import ArgumentABC
|
||||
|
||||
|
||||
class VariableArgument(ArgumentABC):
|
||||
def __init__(
|
||||
self,
|
||||
token: str,
|
||||
name: str,
|
||||
aliases: list[str],
|
||||
value_token: str,
|
||||
prevent_next_executable: bool = False,
|
||||
console_arguments: list["ArgumentABC"] = None,
|
||||
):
|
||||
self._value_token = value_token
|
||||
self._value: str = ""
|
||||
|
||||
ArgumentABC.__init__(self, token, name, aliases, prevent_next_executable, console_arguments)
|
||||
|
||||
@property
|
||||
def value_token(self) -> str:
|
||||
return self._value_token
|
||||
|
||||
@property
|
||||
def value(self) -> str:
|
||||
return self._value
|
||||
|
||||
def set_value(self, value: str):
|
||||
self._value = value
|
||||
@@ -1,5 +1,4 @@
|
||||
from .background_color_enum import BackgroundColorEnum
|
||||
from .console import Console
|
||||
from .console_call import ConsoleCall
|
||||
from ._call import ConsoleCall
|
||||
from .foreground_color_enum import ForegroundColorEnum
|
||||
from .spinner_thread import SpinnerThread
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import multiprocessing
|
||||
import time
|
||||
from multiprocessing import Process
|
||||
|
||||
from termcolor import colored
|
||||
|
||||
@@ -9,8 +10,8 @@ from cpl.core.console.background_color_enum import BackgroundColorEnum
|
||||
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
|
||||
|
||||
|
||||
class SpinnerThread(threading.Thread):
|
||||
r"""Thread to show spinner in terminal
|
||||
class Spinner(Process):
|
||||
r"""Process to show spinner in terminal
|
||||
|
||||
Parameter:
|
||||
msg_len: :class:`int`
|
||||
@@ -22,7 +23,7 @@ class SpinnerThread(threading.Thread):
|
||||
"""
|
||||
|
||||
def __init__(self, msg_len: int, foreground_color: ForegroundColorEnum, background_color: BackgroundColorEnum):
|
||||
threading.Thread.__init__(self)
|
||||
Process.__init__(self)
|
||||
|
||||
self._msg_len = msg_len
|
||||
self._foreground_color = foreground_color
|
||||
@@ -50,29 +51,26 @@ class SpinnerThread(threading.Thread):
|
||||
return color_args
|
||||
|
||||
def run(self) -> None:
|
||||
r"""Entry point of thread, shows the spinner"""
|
||||
r"""Entry point of process, 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()
|
||||
values = os.popen("stty size", "r").read().split()
|
||||
term_rows, term_columns = values if len(values) == 2 else (0, 0)
|
||||
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="")
|
||||
|
||||
padding = columns - self._msg_len - len(end_msg)
|
||||
if padding > 0:
|
||||
print(f'{"" : >{padding}}', 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="")
|
||||
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)):
|
||||
@@ -84,9 +82,10 @@ class SpinnerThread(threading.Thread):
|
||||
if not self._exit:
|
||||
print(colored(end_msg, *self._get_color_args()), end="")
|
||||
|
||||
def stop_spinning(self):
|
||||
def stop(self):
|
||||
r"""Stops the spinner"""
|
||||
self._is_spinning = False
|
||||
super().terminate()
|
||||
time.sleep(0.1)
|
||||
|
||||
def exit(self):
|
||||
@@ -10,9 +10,9 @@ from tabulate import tabulate
|
||||
from termcolor import colored
|
||||
|
||||
from cpl.core.console.background_color_enum import BackgroundColorEnum
|
||||
from cpl.core.console.console_call import ConsoleCall
|
||||
from cpl.core.console._call import ConsoleCall
|
||||
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
|
||||
from cpl.core.console.spinner_thread import SpinnerThread
|
||||
from cpl.core.console._spinner import Spinner
|
||||
|
||||
|
||||
class Console:
|
||||
@@ -464,7 +464,7 @@ class Console:
|
||||
cls.set_hold_back(True)
|
||||
spinner = None
|
||||
if not cls._disabled:
|
||||
spinner = SpinnerThread(len(message), spinner_foreground_color, spinner_background_color)
|
||||
spinner = Spinner(len(message), spinner_foreground_color, spinner_background_color)
|
||||
spinner.start()
|
||||
|
||||
return_value = None
|
||||
@@ -476,7 +476,7 @@ class Console:
|
||||
cls.close()
|
||||
|
||||
if spinner is not None:
|
||||
spinner.stop_spinning()
|
||||
spinner.stop()
|
||||
cls.set_hold_back(False)
|
||||
|
||||
cls.set_foreground_color(ForegroundColorEnum.default)
|
||||
|
||||
@@ -1,3 +0,0 @@
|
||||
from .database_settings_name_enum import DatabaseSettingsNameEnum
|
||||
from .database_settings import DatabaseSettings
|
||||
from .table_abc import TableABC
|
||||
@@ -1,18 +0,0 @@
|
||||
from cpl.core.dependency_injection.scope import Scope
|
||||
from cpl.core.dependency_injection.scope_abc import ScopeABC
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
|
||||
|
||||
class ScopeBuilder:
|
||||
r"""Class to build :class:`cpl.core.dependency_injection.scope.Scope`"""
|
||||
|
||||
def __init__(self, service_provider: ServiceProviderABC) -> None:
|
||||
self._service_provider = service_provider
|
||||
|
||||
def build(self) -> ScopeABC:
|
||||
r"""Returns scope
|
||||
|
||||
Returns:
|
||||
Object of type :class:`cpl.core.dependency_injection.scope.Scope`
|
||||
"""
|
||||
return Scope(self._service_provider)
|
||||
@@ -1,90 +0,0 @@
|
||||
from abc import abstractmethod, ABC
|
||||
from typing import Type
|
||||
|
||||
from cpl.core.database.context.database_context_abc import DatabaseContextABC
|
||||
from cpl.core.database.database_settings import DatabaseSettings
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
from cpl.core.typing import T, Source
|
||||
|
||||
|
||||
class ServiceCollectionABC(ABC):
|
||||
r"""ABC for the class :class:`cpl.core.dependency_injection.service_collection.ServiceCollection`"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def add_db_context(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
|
||||
r"""Adds database context
|
||||
|
||||
Parameter:
|
||||
db_context: Type[:class:`cpl.core.database.context.database_context_abc.DatabaseContextABC`]
|
||||
Database context
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def add_logging(self):
|
||||
r"""Adds the CPL internal logger"""
|
||||
|
||||
@abstractmethod
|
||||
def add_pipes(self):
|
||||
r"""Adds the CPL internal pipes as transient"""
|
||||
|
||||
def add_translation(self):
|
||||
r"""Adds the CPL translation"""
|
||||
raise NotImplementedError("You should install and use the cpl-translation package")
|
||||
|
||||
def add_mail(self):
|
||||
r"""Adds the CPL mail"""
|
||||
raise NotImplementedError("You should install and use the cpl-mail package")
|
||||
|
||||
@abstractmethod
|
||||
def add_transient(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
|
||||
r"""Adds a service with transient lifetime
|
||||
|
||||
Parameter:
|
||||
service_type: :class:`Type`
|
||||
Type of the service
|
||||
service: :class:`Callable`
|
||||
Object of the service
|
||||
|
||||
Returns:
|
||||
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def add_scoped(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
|
||||
r"""Adds a service with scoped lifetime
|
||||
|
||||
Parameter:
|
||||
service_type: :class:`Type`
|
||||
Type of the service
|
||||
service: :class:`Callable`
|
||||
Object of the service
|
||||
|
||||
Returns:
|
||||
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def add_singleton(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
|
||||
r"""Adds a service with singleton lifetime
|
||||
|
||||
Parameter:
|
||||
service_type: :class:`Type`
|
||||
Type of the service
|
||||
service: :class:`Callable`
|
||||
Object of the service
|
||||
|
||||
Returns:
|
||||
self: :class:`cpl.core.dependency_injection.service_collection_abc.ServiceCollectionABC`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def build_service_provider(self) -> ServiceProviderABC:
|
||||
r"""Creates instance of the service provider
|
||||
|
||||
Returns:
|
||||
Object of type :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
|
||||
"""
|
||||
@@ -1,3 +1,2 @@
|
||||
from .application_environment_abc import ApplicationEnvironmentABC
|
||||
from .environment_name_enum import EnvironmentNameEnum
|
||||
from .application_environment import ApplicationEnvironment
|
||||
from .environment_enum import EnvironmentEnum
|
||||
from .environment import Environment
|
||||
|
||||
@@ -1,95 +0,0 @@
|
||||
import os
|
||||
from datetime import datetime
|
||||
from socket import gethostname
|
||||
from typing import Optional
|
||||
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
from cpl.core.environment.environment_name_enum import EnvironmentNameEnum
|
||||
|
||||
|
||||
class ApplicationEnvironment(ApplicationEnvironmentABC):
|
||||
r"""Represents environment of the application
|
||||
|
||||
Parameter:
|
||||
name: :class:`cpl.core.environment.environment_name_enum.EnvironmentNameEnum`
|
||||
"""
|
||||
|
||||
def __init__(self, name: EnvironmentNameEnum = EnvironmentNameEnum.production):
|
||||
ApplicationEnvironmentABC.__init__(self)
|
||||
|
||||
self._environment_name: Optional[EnvironmentNameEnum] = name
|
||||
self._app_name: Optional[str] = None
|
||||
self._customer: Optional[str] = None
|
||||
|
||||
self._start_time: datetime = datetime.now()
|
||||
self._end_time: datetime = datetime.now()
|
||||
self._runtime_directory = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
self._working_directory = os.getcwd()
|
||||
|
||||
@property
|
||||
def environment_name(self) -> str:
|
||||
return str(self._environment_name.value)
|
||||
|
||||
@environment_name.setter
|
||||
def environment_name(self, environment_name: str):
|
||||
self._environment_name = EnvironmentNameEnum(environment_name)
|
||||
|
||||
@property
|
||||
def application_name(self) -> str:
|
||||
return self._app_name if self._app_name is not None else ""
|
||||
|
||||
@application_name.setter
|
||||
def application_name(self, application_name: str):
|
||||
self._app_name = application_name
|
||||
|
||||
@property
|
||||
def customer(self) -> str:
|
||||
return self._customer if self._customer is not None else ""
|
||||
|
||||
@customer.setter
|
||||
def customer(self, customer: str):
|
||||
self._customer = customer
|
||||
|
||||
@property
|
||||
def host_name(self):
|
||||
return gethostname()
|
||||
|
||||
@property
|
||||
def start_time(self) -> datetime:
|
||||
return self._start_time
|
||||
|
||||
@property
|
||||
def end_time(self) -> datetime:
|
||||
return self._end_time
|
||||
|
||||
@end_time.setter
|
||||
def end_time(self, end_time: datetime):
|
||||
self._end_time = end_time
|
||||
|
||||
@property
|
||||
def date_time_now(self) -> datetime:
|
||||
return datetime.now()
|
||||
|
||||
@property
|
||||
def working_directory(self) -> str:
|
||||
return str(self._working_directory)
|
||||
|
||||
@property
|
||||
def runtime_directory(self) -> str:
|
||||
return str(self._runtime_directory)
|
||||
|
||||
def set_runtime_directory(self, runtime_directory: str):
|
||||
if runtime_directory != "":
|
||||
self._runtime_directory = runtime_directory
|
||||
return
|
||||
|
||||
self._runtime_directory = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def set_working_directory(self, working_directory: str):
|
||||
if working_directory != "":
|
||||
self._working_directory = working_directory
|
||||
os.chdir(self._working_directory)
|
||||
return
|
||||
|
||||
self._working_directory = os.path.abspath("./")
|
||||
os.chdir(self._working_directory)
|
||||
@@ -1,98 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class ApplicationEnvironmentABC(ABC):
|
||||
r"""ABC of the class :class:`cpl.core.environment.application_environment.ApplicationEnvironment`"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def environment_name(self) -> str:
|
||||
pass
|
||||
|
||||
@environment_name.setter
|
||||
@abstractmethod
|
||||
def environment_name(self, environment_name: str):
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def application_name(self) -> str:
|
||||
pass
|
||||
|
||||
@application_name.setter
|
||||
@abstractmethod
|
||||
def application_name(self, application_name: str):
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def customer(self) -> str:
|
||||
pass
|
||||
|
||||
@customer.setter
|
||||
@abstractmethod
|
||||
def customer(self, customer: str):
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def host_name(self) -> str:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def start_time(self) -> datetime:
|
||||
pass
|
||||
|
||||
@start_time.setter
|
||||
@abstractmethod
|
||||
def start_time(self, start_time: datetime):
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def end_time(self):
|
||||
pass
|
||||
|
||||
@end_time.setter
|
||||
@abstractmethod
|
||||
def end_time(self, end_time: datetime):
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def date_time_now(self) -> datetime:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def working_directory(self) -> str:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def runtime_directory(self) -> str:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def set_runtime_directory(self, runtime_directory: str):
|
||||
r"""Sets the current runtime directory
|
||||
|
||||
Parameter:
|
||||
runtime_directory: :class:`str`
|
||||
Path of the runtime directory
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def set_working_directory(self, working_directory: str):
|
||||
r"""Sets the current working directory
|
||||
|
||||
Parameter:
|
||||
working_directory: :class:`str`
|
||||
Path of the current working directory
|
||||
"""
|
||||
@@ -1,22 +1,58 @@
|
||||
import os
|
||||
from socket import gethostname
|
||||
from typing import Optional, Type
|
||||
|
||||
from cpl.core.environment.environment_enum import EnvironmentEnum
|
||||
from cpl.core.typing import T
|
||||
from cpl.core.utils.get_value import get_value
|
||||
|
||||
|
||||
class Environment:
|
||||
_environment = "production"
|
||||
r"""Represents environment of the application
|
||||
|
||||
Parameter:
|
||||
name: :class:`cpl.core.environment.environment_name_enum.EnvironmentNameEnum`
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def get_environment(cls):
|
||||
return cls._environment
|
||||
return cls.get("ENVIRONMENT", str, EnvironmentEnum.production.value)
|
||||
|
||||
@classmethod
|
||||
def set_environment(cls, environment: str):
|
||||
if environment not in ["development", "staging", "production"]:
|
||||
raise ValueError("Invalid environment")
|
||||
Environment._environment = environment
|
||||
assert environment is not None and environment != "", "environment must not be None or empty"
|
||||
assert environment.lower() in [
|
||||
e.value for e in EnvironmentEnum
|
||||
], f"environment must be one of {[e.value for e in EnvironmentEnum]}"
|
||||
cls.set("ENVIRONMENT", environment.lower())
|
||||
|
||||
@classmethod
|
||||
def get_app_name(cls) -> str:
|
||||
return cls.get("APP_NAME", str)
|
||||
|
||||
@classmethod
|
||||
def set_app_name(cls, app_name: str):
|
||||
cls.set("APP_NAME", app_name)
|
||||
|
||||
@staticmethod
|
||||
def get_host_name() -> str:
|
||||
return gethostname()
|
||||
|
||||
@staticmethod
|
||||
def get_cwd() -> str:
|
||||
return os.getcwd()
|
||||
|
||||
@staticmethod
|
||||
def set_cwd(working_directory: str):
|
||||
assert working_directory is not None and working_directory != "", "working_directory must not be None or empty"
|
||||
|
||||
os.chdir(working_directory)
|
||||
|
||||
@staticmethod
|
||||
def set(key: str, value: T):
|
||||
assert key is not None and key != "", "key must not be None or empty"
|
||||
|
||||
os.environ[key] = str(value)
|
||||
|
||||
@staticmethod
|
||||
def get(key: str, cast_type: Type[T], default: Optional[T] = None) -> Optional[T]:
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class EnvironmentNameEnum(Enum):
|
||||
class EnvironmentEnum(Enum):
|
||||
production = "production"
|
||||
staging = "staging"
|
||||
testing = "testing"
|
||||
@@ -1,94 +0,0 @@
|
||||
import multiprocessing
|
||||
import os
|
||||
from datetime import datetime
|
||||
from typing import Self
|
||||
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.log.log_level_enum import LogLevelEnum
|
||||
|
||||
|
||||
class LogWriter:
|
||||
_instance = None
|
||||
|
||||
# ANSI color codes for different log levels
|
||||
_COLORS = {
|
||||
LogLevelEnum.trace: "\033[37m", # Light Gray
|
||||
LogLevelEnum.debug: "\033[94m", # Blue
|
||||
LogLevelEnum.info: "\033[92m", # Green
|
||||
LogLevelEnum.warning: "\033[93m", # Yellow
|
||||
LogLevelEnum.error: "\033[91m", # Red
|
||||
LogLevelEnum.fatal: "\033[95m", # Magenta
|
||||
}
|
||||
|
||||
def __init__(self, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info):
|
||||
self._file_prefix = file_prefix
|
||||
self._level = level
|
||||
|
||||
self._queue = multiprocessing.Queue()
|
||||
self._process = multiprocessing.Process(target=self._log_worker, daemon=True)
|
||||
|
||||
self._create_log_dir()
|
||||
self._process.start()
|
||||
|
||||
@property
|
||||
def level(self) -> LogLevelEnum:
|
||||
return self._level
|
||||
|
||||
@level.setter
|
||||
def level(self, value: LogLevelEnum):
|
||||
assert isinstance(value, LogLevelEnum), "Log level must be an instance of LogLevelEnum"
|
||||
self._level = value
|
||||
|
||||
@classmethod
|
||||
def get_instance(cls, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info) -> Self:
|
||||
if cls._instance is None:
|
||||
cls._instance = LogWriter(file_prefix, level)
|
||||
return cls._instance
|
||||
|
||||
@staticmethod
|
||||
def _create_log_dir():
|
||||
if os.path.exists("logs"):
|
||||
return
|
||||
|
||||
os.makedirs("logs")
|
||||
|
||||
def _log_worker(self):
|
||||
"""Worker process that writes log messages from the queue to the file."""
|
||||
while True:
|
||||
content = self._queue.get()
|
||||
if content is None: # Shutdown signal
|
||||
break
|
||||
self._write_log_to_file(content)
|
||||
Console.write_line(
|
||||
f"{self._COLORS.get(self._level, '\033[0m')}{content}\033[0m"
|
||||
)
|
||||
|
||||
@property
|
||||
def log_file(self):
|
||||
return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
|
||||
|
||||
def _ensure_file_size(self):
|
||||
log_file = self.log_file
|
||||
if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
|
||||
return
|
||||
|
||||
# if exists and size is greater than 300MB, create a new file
|
||||
os.rename(
|
||||
log_file,
|
||||
f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
|
||||
)
|
||||
|
||||
def _write_log_to_file(self, content: str):
|
||||
self._ensure_file_size()
|
||||
with open(self.log_file, "a") as log_file:
|
||||
log_file.write(content + "\n")
|
||||
log_file.close()
|
||||
|
||||
def log(self, content: str):
|
||||
"""Enqueue log message without blocking main app."""
|
||||
self._queue.put(content)
|
||||
|
||||
def close(self):
|
||||
"""Gracefully stop the logging process."""
|
||||
self._queue.put(None)
|
||||
self._process.join()
|
||||
@@ -3,7 +3,6 @@ import traceback
|
||||
from datetime import datetime
|
||||
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.log._log_writer import LogWriter
|
||||
from cpl.core.log.log_level_enum import LogLevelEnum
|
||||
from cpl.core.log.logger_abc import LoggerABC
|
||||
from cpl.core.typing import Messages, Source
|
||||
@@ -13,6 +12,16 @@ class Logger(LoggerABC):
|
||||
_level = LogLevelEnum.info
|
||||
_levels = [x for x in LogLevelEnum]
|
||||
|
||||
# ANSI color codes for different log levels
|
||||
_COLORS = {
|
||||
LogLevelEnum.trace: "\033[37m", # Light Gray
|
||||
LogLevelEnum.debug: "\033[94m", # Blue
|
||||
LogLevelEnum.info: "\033[92m", # Green
|
||||
LogLevelEnum.warning: "\033[93m", # Yellow
|
||||
LogLevelEnum.error: "\033[91m", # Red
|
||||
LogLevelEnum.fatal: "\033[95m", # Magenta
|
||||
}
|
||||
|
||||
def __init__(self, source: Source, file_prefix: str = None):
|
||||
LoggerABC.__init__(self)
|
||||
assert source is not None and source != "", "Source cannot be None or empty"
|
||||
@@ -22,7 +31,18 @@ class Logger(LoggerABC):
|
||||
file_prefix = "app"
|
||||
|
||||
self._file_prefix = file_prefix
|
||||
self._writer = LogWriter.get_instance(self._file_prefix)
|
||||
self._create_log_dir()
|
||||
|
||||
@property
|
||||
def log_file(self):
|
||||
return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
|
||||
|
||||
@staticmethod
|
||||
def _create_log_dir():
|
||||
if os.path.exists("logs"):
|
||||
return
|
||||
|
||||
os.makedirs("logs")
|
||||
|
||||
@classmethod
|
||||
def set_level(cls, level: LogLevelEnum):
|
||||
@@ -31,6 +51,24 @@ class Logger(LoggerABC):
|
||||
else:
|
||||
raise ValueError(f"Invalid log level: {level}")
|
||||
|
||||
@staticmethod
|
||||
def _ensure_file_size(log_file: str):
|
||||
if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
|
||||
return
|
||||
|
||||
# if exists and size is greater than 300MB, create a new file
|
||||
os.rename(
|
||||
log_file,
|
||||
f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
|
||||
)
|
||||
|
||||
def _write_log_to_file(self, content: str):
|
||||
file = self.log_file
|
||||
self._ensure_file_size(file)
|
||||
with open(file, "a") as log_file:
|
||||
log_file.write(content + "\n")
|
||||
log_file.close()
|
||||
|
||||
def _log(self, level: LogLevelEnum, *messages: Messages):
|
||||
try:
|
||||
if self._levels.index(level) < self._levels.index(self._level):
|
||||
@@ -39,7 +77,8 @@ class Logger(LoggerABC):
|
||||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
|
||||
formatted_message = self._format_message(level.value, timestamp, *messages)
|
||||
|
||||
self._writer.log(formatted_message)
|
||||
self._write_log_to_file(formatted_message)
|
||||
Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{formatted_message}\033[0m")
|
||||
except Exception as e:
|
||||
print(f"Error while logging: {e} -> {traceback.format_exc()}")
|
||||
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
from cpl.core.pipes.pipe_abc import PipeABC
|
||||
from cpl.core.typing import T
|
||||
|
||||
|
||||
class BoolPipe(PipeABC):
|
||||
def __init__(self):
|
||||
pass
|
||||
class BoolPipe[bool](PipeABC):
|
||||
|
||||
def transform(self, value: bool, *args):
|
||||
return "True" if value else "False"
|
||||
@staticmethod
|
||||
def to_str(value: T, *args):
|
||||
return str(value).lower()
|
||||
|
||||
@staticmethod
|
||||
def from_str(value: str, *args) -> T:
|
||||
return value in ("True", "true", "1", "yes", "y", "Y")
|
||||
|
||||
@@ -1,20 +1,19 @@
|
||||
from cpl.core.pipes.pipe_abc import PipeABC
|
||||
from cpl.core.typing import T
|
||||
|
||||
|
||||
class IPAddressPipe(PipeABC):
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def transform(self, value: list[int], *args):
|
||||
class IPAddressPipe[list](PipeABC):
|
||||
@staticmethod
|
||||
def to_str(value: T, *args) -> str:
|
||||
string = ""
|
||||
|
||||
if len(value) != 4:
|
||||
raise Exception("Invalid IP")
|
||||
raise ValueError("Invalid IP")
|
||||
|
||||
for i in range(0, len(value)):
|
||||
byte = value[i]
|
||||
if byte > 255 or byte < 0:
|
||||
raise Exception("Invalid IP")
|
||||
if not 0 <= byte <= 255:
|
||||
raise ValueError("Invalid IP")
|
||||
|
||||
if i == len(value) - 1:
|
||||
string += f"{byte}"
|
||||
@@ -22,3 +21,18 @@ class IPAddressPipe(PipeABC):
|
||||
string += f"{byte}."
|
||||
|
||||
return string
|
||||
|
||||
@staticmethod
|
||||
def from_str(value: str, *args) -> T:
|
||||
parts = value.split(".")
|
||||
if len(parts) != 4:
|
||||
raise Exception("Invalid IP")
|
||||
|
||||
result = []
|
||||
for part in parts:
|
||||
byte = int(part)
|
||||
if not 0 <= byte <= 255:
|
||||
raise Exception("Invalid IP")
|
||||
result.append(byte)
|
||||
|
||||
return result
|
||||
|
||||
@@ -1,11 +1,16 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Generic
|
||||
|
||||
from cpl.core.typing import T
|
||||
|
||||
|
||||
class PipeABC(ABC):
|
||||
class PipeABC(ABC, Generic[T]):
|
||||
@staticmethod
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
def to_str(value: T, *args) -> str:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
@abstractmethod
|
||||
def transform(self, value: any, *args):
|
||||
def from_str(value: str, *args) -> T:
|
||||
pass
|
||||
|
||||
@@ -1,17 +0,0 @@
|
||||
from cpl_cli.configuration import VersionSettingsNameEnum
|
||||
from cpl.core.pipes.pipe_abc import PipeABC
|
||||
|
||||
|
||||
class VersionPipe(PipeABC):
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def transform(self, value: dict, *args):
|
||||
for atr in VersionSettingsNameEnum:
|
||||
if atr.value not in value:
|
||||
raise KeyError(atr.value)
|
||||
|
||||
v_str = f"{value[VersionSettingsNameEnum.major.value]}.{value[VersionSettingsNameEnum.minor.value]}"
|
||||
if value[VersionSettingsNameEnum.micro.value] is not None:
|
||||
v_str += f".{value[VersionSettingsNameEnum.micro.value]}"
|
||||
return v_str
|
||||
@@ -1,6 +1,7 @@
|
||||
from typing import TypeVar, Any
|
||||
|
||||
T = TypeVar("T")
|
||||
D = TypeVar("D")
|
||||
R = TypeVar("R")
|
||||
|
||||
Service = TypeVar("Service")
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from .b64 import B64
|
||||
from .credential_manager import CredentialManager
|
||||
from .string import String
|
||||
from .json_processor import JSONProcessor
|
||||
from .pip import Pip
|
||||
from .string import String
|
||||
|
||||
43
src/cpl-core/cpl/core/utils/b64.py
Normal file
43
src/cpl-core/cpl/core/utils/b64.py
Normal file
@@ -0,0 +1,43 @@
|
||||
import base64
|
||||
from typing import Union
|
||||
|
||||
|
||||
class B64:
|
||||
|
||||
@staticmethod
|
||||
def encode(string: str) -> str:
|
||||
"""
|
||||
Encode a string with base64
|
||||
:param string:
|
||||
:return:
|
||||
"""
|
||||
return base64.b64encode(string.encode("utf-8")).decode("utf-8")
|
||||
|
||||
@staticmethod
|
||||
def decode(string: str) -> str:
|
||||
"""
|
||||
Decode a string with base64
|
||||
:param string:
|
||||
:return:
|
||||
"""
|
||||
return base64.b64decode(string).decode("utf-8")
|
||||
|
||||
@staticmethod
|
||||
def is_b64(sb: Union[str, bytes]) -> bool:
|
||||
"""
|
||||
Check if a string is base64 encoded
|
||||
:param Union[str, bytes] sb:
|
||||
:return:
|
||||
:rtype: bool
|
||||
"""
|
||||
try:
|
||||
if isinstance(sb, str):
|
||||
# If there's any unicode here, an exception will be thrown and the function will return false
|
||||
sb_bytes = bytes(sb, "ascii")
|
||||
elif isinstance(sb, bytes):
|
||||
sb_bytes = sb
|
||||
else:
|
||||
raise ValueError("Argument must be string or bytes")
|
||||
return base64.b64encode(base64.b64decode(sb_bytes)) == sb_bytes
|
||||
except ValueError:
|
||||
return False
|
||||
@@ -40,16 +40,9 @@ def get_value(
|
||||
if cast_type == bool:
|
||||
return value.lower() in ["true", "1"]
|
||||
|
||||
if (
|
||||
cast_type if not hasattr(cast_type, "__origin__") else cast_type.__origin__
|
||||
) == list:
|
||||
if (
|
||||
not (value.startswith("[") and value.endswith("]"))
|
||||
and list_delimiter not in value
|
||||
):
|
||||
raise ValueError(
|
||||
"List values must be enclosed in square brackets or use a delimiter."
|
||||
)
|
||||
if (cast_type if not hasattr(cast_type, "__origin__") else cast_type.__origin__) == list:
|
||||
if not (value.startswith("[") and value.endswith("]")) and list_delimiter not in value:
|
||||
raise ValueError("List values must be enclosed in square brackets or use a delimiter.")
|
||||
|
||||
if value.startswith("[") and value.endswith("]"):
|
||||
value = value[1:-1]
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import enum
|
||||
from inspect import signature, Parameter
|
||||
|
||||
from cpl.core.utils import String
|
||||
from cpl.core.utils.string import String
|
||||
|
||||
|
||||
class JSONProcessor:
|
||||
@@ -16,7 +16,7 @@ class JSONProcessor:
|
||||
if parameter.name == "self" or parameter.annotation == Parameter.empty:
|
||||
continue
|
||||
|
||||
name = String.first_to_upper(String.convert_to_camel_case(parameter.name))
|
||||
name = String.first_to_upper(String.to_camel_case(parameter.name))
|
||||
name_first_lower = String.first_to_lower(name)
|
||||
if name in values or name_first_lower in values or name.upper() in values:
|
||||
value = ""
|
||||
|
||||
@@ -1,13 +1,13 @@
|
||||
import random
|
||||
import re
|
||||
import string
|
||||
import random
|
||||
|
||||
|
||||
class String:
|
||||
r"""Useful functions for strings"""
|
||||
|
||||
@staticmethod
|
||||
def convert_to_camel_case(chars: str) -> str:
|
||||
def to_camel_case(s: str) -> str:
|
||||
r"""Converts string to camel case
|
||||
|
||||
Parameter:
|
||||
@@ -17,16 +17,10 @@ class String:
|
||||
Returns:
|
||||
String converted to CamelCase
|
||||
"""
|
||||
converted_name = chars
|
||||
char_set = string.punctuation + " "
|
||||
for char in char_set:
|
||||
if char in converted_name:
|
||||
converted_name = "".join(word.title() for word in converted_name.split(char))
|
||||
|
||||
return converted_name
|
||||
return re.sub(r"(?<!^)(?=[A-Z])", "_", s).lower()
|
||||
|
||||
@staticmethod
|
||||
def convert_to_snake_case(chars: str) -> str:
|
||||
def to_snake_case(chars: str) -> str:
|
||||
r"""Converts string to snake case
|
||||
|
||||
Parameter:
|
||||
@@ -56,7 +50,7 @@ class String:
|
||||
return re.sub(pattern2, r"\1_\2", file_name).lower()
|
||||
|
||||
@staticmethod
|
||||
def first_to_upper(chars: str) -> str:
|
||||
def first_to_upper(s: str) -> str:
|
||||
r"""Converts first char to upper
|
||||
|
||||
Parameter:
|
||||
@@ -66,10 +60,10 @@ class String:
|
||||
Returns:
|
||||
String with first char as upper
|
||||
"""
|
||||
return f"{chars[0].upper()}{chars[1:]}"
|
||||
return s[0].upper() + s[1:] if s else s
|
||||
|
||||
@staticmethod
|
||||
def first_to_lower(chars: str) -> str:
|
||||
def first_to_lower(s: str) -> str:
|
||||
r"""Converts first char to lower
|
||||
|
||||
Parameter:
|
||||
@@ -79,14 +73,24 @@ class String:
|
||||
Returns:
|
||||
String with first char as lower
|
||||
"""
|
||||
return f"{chars[0].lower()}{chars[1:]}"
|
||||
return s[0].lower() + s[1:] if s else s
|
||||
|
||||
@staticmethod
|
||||
def random_string(chars: str, length: int) -> str:
|
||||
def random(length: int, letters=True, digits=False, special_characters=False) -> str:
|
||||
r"""Creates random string by given chars and length
|
||||
|
||||
Returns:
|
||||
String of random chars
|
||||
"""
|
||||
|
||||
return "".join(random.choice(chars) for _ in range(length))
|
||||
characters = []
|
||||
if letters:
|
||||
characters.append(string.ascii_letters)
|
||||
|
||||
if digits:
|
||||
characters.append(string.digits)
|
||||
|
||||
if special_characters:
|
||||
characters.append(string.punctuation)
|
||||
|
||||
return "".join(random.choice(characters) for _ in range(length)) if characters else ""
|
||||
|
||||
@@ -3,3 +3,4 @@ colorama==0.4.6
|
||||
tabulate==0.9.0
|
||||
termcolor==3.1.0
|
||||
mysql-connector-python==9.4.0
|
||||
pynput==1.8.1
|
||||
|
||||
23
src/cpl-database/cpl/database/__init__.py
Normal file
23
src/cpl-database/cpl/database/__init__.py
Normal file
@@ -0,0 +1,23 @@
|
||||
from cpl.dependency import ServiceCollection as _ServiceCollection
|
||||
from . import mysql
|
||||
from .database_settings import DatabaseSettings
|
||||
from .database_settings_name_enum import DatabaseSettingsNameEnum
|
||||
from .mysql.context import DatabaseContextABC, DatabaseContext
|
||||
from .table_abc import TableABC
|
||||
|
||||
|
||||
def add_mysql(collection: _ServiceCollection):
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.configuration import Configuration
|
||||
|
||||
try:
|
||||
collection.add_singleton(DatabaseContextABC, DatabaseContext)
|
||||
database_context = collection.build_service_provider().get_service(DatabaseContextABC)
|
||||
|
||||
db_settings: DatabaseSettings = Configuration.get(DatabaseSettings)
|
||||
database_context.connect(db_settings)
|
||||
except ImportError as e:
|
||||
Console.error("cpl-translation is not installed", str(e))
|
||||
|
||||
|
||||
_ServiceCollection.with_module(add_mysql, mysql.__name__)
|
||||
@@ -5,4 +5,4 @@ from cpl.core.typing import Source
|
||||
class DBLogger(Logger):
|
||||
|
||||
def __init__(self, source: Source):
|
||||
Logger.__init__(self, source, "db")
|
||||
Logger.__init__(self, source, "db")
|
||||
@@ -4,8 +4,8 @@ import mysql.connector as sql
|
||||
from mysql.connector.abstracts import MySQLConnectionAbstract
|
||||
from mysql.connector.cursor import MySQLCursorBuffered
|
||||
|
||||
from cpl.core.database.connection.database_connection_abc import DatabaseConnectionABC
|
||||
from cpl.core.database.database_settings import DatabaseSettings
|
||||
from cpl.database.mysql.connection.database_connection_abc import DatabaseConnectionABC
|
||||
from cpl.database.database_settings import DatabaseSettings
|
||||
from cpl.core.utils.credential_manager import CredentialManager
|
||||
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.database.database_settings import DatabaseSettings
|
||||
from cpl.database.database_settings import DatabaseSettings
|
||||
from mysql.connector.abstracts import MySQLConnectionAbstract
|
||||
from mysql.connector.cursor import MySQLCursorBuffered
|
||||
|
||||
|
||||
class DatabaseConnectionABC(ABC):
|
||||
r"""ABC for the :class:`cpl.core.database.connection.database_connection.DatabaseConnection`"""
|
||||
r"""ABC for the :class:`cpl.database.connection.database_connection.DatabaseConnection`"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self):
|
||||
@@ -1,10 +1,10 @@
|
||||
from typing import Optional
|
||||
|
||||
|
||||
from cpl.core.database.connection.database_connection import DatabaseConnection
|
||||
from cpl.core.database.connection.database_connection_abc import DatabaseConnectionABC
|
||||
from cpl.core.database.context.database_context_abc import DatabaseContextABC
|
||||
from cpl.core.database.database_settings import DatabaseSettings
|
||||
from cpl.database.mysql.connection.database_connection import DatabaseConnection
|
||||
from cpl.database.mysql.connection.database_connection_abc import DatabaseConnectionABC
|
||||
from cpl.database.mysql.context.database_context_abc import DatabaseContextABC
|
||||
from cpl.database.database_settings import DatabaseSettings
|
||||
from mysql.connector.cursor import MySQLCursorBuffered
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@ class DatabaseContext(DatabaseContextABC):
|
||||
r"""Representation of the database context
|
||||
|
||||
Parameter:
|
||||
database_settings: :class:`cpl.core.database.database_settings.DatabaseSettings`
|
||||
database_settings: :class:`cpl.database.database_settings.DatabaseSettings`
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
@@ -1,11 +1,11 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from cpl.core.database.database_settings import DatabaseSettings
|
||||
from cpl.database.database_settings import DatabaseSettings
|
||||
from mysql.connector.cursor import MySQLCursorBuffered
|
||||
|
||||
|
||||
class DatabaseContextABC(ABC):
|
||||
r"""ABC for the :class:`cpl.core.database.context.database_context.DatabaseContext`"""
|
||||
r"""ABC for the :class:`cpl.database.context.database_context.DatabaseContext`"""
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self, *args):
|
||||
@@ -21,7 +21,7 @@ class DatabaseContextABC(ABC):
|
||||
r"""Connects to a database by connection settings
|
||||
|
||||
Parameter:
|
||||
database_settings :class:`cpl.core.database.database_settings.DatabaseSettings`
|
||||
database_settings :class:`cpl.database.database_settings.DatabaseSettings`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
30
src/cpl-database/pyproject.toml
Normal file
30
src/cpl-database/pyproject.toml
Normal file
@@ -0,0 +1,30 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "cpl-database"
|
||||
version = "2024.7.0"
|
||||
description = "CPL database"
|
||||
readme ="CPL database package"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
authors = [
|
||||
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
|
||||
]
|
||||
keywords = ["cpl", "database", "backend", "shared", "library"]
|
||||
|
||||
dynamic = ["dependencies", "optional-dependencies"]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://www.sh-edraft.de"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
include = ["cpl*"]
|
||||
|
||||
[tool.setuptools.dynamic]
|
||||
dependencies = { file = ["requirements.txt"] }
|
||||
optional-dependencies.dev = { file = ["requirements.dev.txt"] }
|
||||
|
||||
|
||||
1
src/cpl-database/requirements.dev.txt
Normal file
1
src/cpl-database/requirements.dev.txt
Normal file
@@ -0,0 +1 @@
|
||||
black==25.1.0
|
||||
2
src/cpl-database/requirements.txt
Normal file
2
src/cpl-database/requirements.txt
Normal file
@@ -0,0 +1,2 @@
|
||||
cpl-core
|
||||
cpl-dependency
|
||||
@@ -1,7 +1,6 @@
|
||||
from .scope import Scope
|
||||
from .scope_abc import ScopeABC
|
||||
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
|
||||
@@ -1,5 +1,5 @@
|
||||
from cpl.core.dependency_injection.scope_abc import ScopeABC
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
from cpl.dependency.scope_abc import ScopeABC
|
||||
from cpl.dependency.service_provider_abc import ServiceProviderABC
|
||||
|
||||
|
||||
class Scope(ScopeABC):
|
||||
@@ -2,7 +2,7 @@ from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class ScopeABC(ABC):
|
||||
r"""ABC for the class :class:`cpl.core.dependency_injection.scope.Scope`"""
|
||||
r"""ABC for the class :class:`cpl.dependency.scope.Scope`"""
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
@@ -13,7 +13,7 @@ class ScopeABC(ABC):
|
||||
r"""Returns to service provider of scope
|
||||
|
||||
Returns:
|
||||
Object of type :class:`cpl.core.dependency_injection.service_provider_abc.ServiceProviderABC`
|
||||
Object of type :class:`cpl.dependency.service_provider_abc.ServiceProviderABC`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
18
src/cpl-dependency/cpl/dependency/scope_builder.py
Normal file
18
src/cpl-dependency/cpl/dependency/scope_builder.py
Normal file
@@ -0,0 +1,18 @@
|
||||
from cpl.dependency.scope import Scope
|
||||
from cpl.dependency.scope_abc import ScopeABC
|
||||
from cpl.dependency.service_provider_abc import ServiceProviderABC
|
||||
|
||||
|
||||
class ScopeBuilder:
|
||||
r"""Class to build :class:`cpl.dependency.scope.Scope`"""
|
||||
|
||||
def __init__(self, service_provider: ServiceProviderABC) -> None:
|
||||
self._service_provider = service_provider
|
||||
|
||||
def build(self) -> ScopeABC:
|
||||
r"""Returns scope
|
||||
|
||||
Returns:
|
||||
Object of type :class:`cpl.dependency.scope.Scope`
|
||||
"""
|
||||
return Scope(self._service_provider)
|
||||
@@ -1,27 +1,26 @@
|
||||
from typing import Union, Type, Callable, Optional
|
||||
|
||||
from cpl.core.configuration.configuration_abc import ConfigurationABC
|
||||
from cpl.core.database.context.database_context_abc import DatabaseContextABC
|
||||
from cpl.core.database.database_settings import DatabaseSettings
|
||||
from cpl.core.dependency_injection.service_collection_abc import ServiceCollectionABC
|
||||
from cpl.core.dependency_injection.service_descriptor import ServiceDescriptor
|
||||
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
|
||||
from cpl.core.dependency_injection.service_provider import ServiceProvider
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
from cpl.core.log.logger import Logger
|
||||
from cpl.core.log.logger_abc import LoggerABC
|
||||
from cpl.core.pipes.pipe_abc import PipeABC
|
||||
from cpl.core.typing import T, Service
|
||||
from cpl.dependency.service_descriptor import ServiceDescriptor
|
||||
from cpl.dependency.service_lifetime_enum import ServiceLifetimeEnum
|
||||
from cpl.dependency.service_provider import ServiceProvider
|
||||
from cpl.dependency.service_provider_abc import ServiceProviderABC
|
||||
|
||||
|
||||
class ServiceCollection(ServiceCollectionABC):
|
||||
class ServiceCollection:
|
||||
r"""Representation of the collection of services"""
|
||||
|
||||
def __init__(self, config: ConfigurationABC):
|
||||
ServiceCollectionABC.__init__(self)
|
||||
self._configuration: ConfigurationABC = config
|
||||
_modules: dict[str, Callable] = {}
|
||||
|
||||
self._database_context: Optional[DatabaseContextABC] = None
|
||||
@classmethod
|
||||
def with_module(cls, func: Callable, name: str = None):
|
||||
cls._modules[func.__name__ if name is None else name] = func
|
||||
return cls
|
||||
|
||||
def __init__(self):
|
||||
self._service_descriptors: list[ServiceDescriptor] = []
|
||||
|
||||
def _add_descriptor(self, service: Union[type, object], lifetime: ServiceLifetimeEnum, base_type: Callable = None):
|
||||
@@ -47,10 +46,19 @@ class ServiceCollection(ServiceCollectionABC):
|
||||
|
||||
return self
|
||||
|
||||
def add_db_context(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
|
||||
self.add_singleton(DatabaseContextABC, db_context_type)
|
||||
self._database_context = self.build_service_provider().get_service(DatabaseContextABC)
|
||||
self._database_context.connect(db_settings)
|
||||
def add_module(self, module: str | object):
|
||||
if not isinstance(module, str):
|
||||
module = module.__name__
|
||||
|
||||
if module not in self._modules:
|
||||
raise ValueError(f"Module {module} not found")
|
||||
|
||||
self._modules[module](self)
|
||||
|
||||
# def add_mysql(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
|
||||
# self.add_singleton(DatabaseContextABC, db_context_type)
|
||||
# self._database_context = self.build_service_provider().get_service(DatabaseContextABC)
|
||||
# self._database_context.connect(db_settings)
|
||||
|
||||
def add_logging(self):
|
||||
self.add_transient(LoggerABC, Logger)
|
||||
@@ -74,6 +82,6 @@ class ServiceCollection(ServiceCollectionABC):
|
||||
return self
|
||||
|
||||
def build_service_provider(self) -> ServiceProviderABC:
|
||||
sp = ServiceProvider(self._service_descriptors, self._configuration, self._database_context)
|
||||
sp = ServiceProvider(self._service_descriptors)
|
||||
ServiceProviderABC.set_global_provider(sp)
|
||||
return sp
|
||||
@@ -1,6 +1,6 @@
|
||||
from typing import Union, Optional
|
||||
|
||||
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
|
||||
from cpl.dependency.service_lifetime_enum import ServiceLifetimeEnum
|
||||
|
||||
|
||||
class ServiceDescriptor:
|
||||
@@ -9,7 +9,7 @@ class ServiceDescriptor:
|
||||
Parameter:
|
||||
implementation: Union[:class:`type`, Optional[:class:`object`]]
|
||||
Object or type of service
|
||||
lifetime: :class:`cpl.core.dependency_injection.service_lifetime_enum.ServiceLifetimeEnum`
|
||||
lifetime: :class:`cpl.dependency.service_lifetime_enum.ServiceLifetimeEnum`
|
||||
Lifetime of the service
|
||||
"""
|
||||
|
||||
@@ -3,16 +3,15 @@ import typing
|
||||
from inspect import signature, Parameter, Signature
|
||||
from typing import Optional
|
||||
|
||||
from cpl.core.configuration.configuration_abc import ConfigurationABC
|
||||
from cpl.core.configuration import Configuration
|
||||
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
|
||||
from cpl.core.database.context.database_context_abc import DatabaseContextABC
|
||||
from cpl.core.dependency_injection.scope_abc import ScopeABC
|
||||
from cpl.core.dependency_injection.scope_builder import ScopeBuilder
|
||||
from cpl.core.dependency_injection.service_descriptor import ServiceDescriptor
|
||||
from cpl.core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
|
||||
from cpl.core.dependency_injection.service_provider_abc import ServiceProviderABC
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
from cpl.core.environment import Environment
|
||||
from cpl.core.typing import T, R, Source
|
||||
from cpl.dependency.scope_abc import ScopeABC
|
||||
from cpl.dependency.scope_builder import ScopeBuilder
|
||||
from cpl.dependency.service_descriptor import ServiceDescriptor
|
||||
from cpl.dependency.service_lifetime_enum import ServiceLifetimeEnum
|
||||
from cpl.dependency.service_provider_abc import ServiceProviderABC
|
||||
|
||||
|
||||
class ServiceProvider(ServiceProviderABC):
|
||||
@@ -20,25 +19,21 @@ class ServiceProvider(ServiceProviderABC):
|
||||
|
||||
Parameter
|
||||
---------
|
||||
service_descriptors: list[:class:`cpl.core.dependency_injection.service_descriptor.ServiceDescriptor`]
|
||||
service_descriptors: list[:class:`cpl.dependency.service_descriptor.ServiceDescriptor`]
|
||||
Descriptor of the service
|
||||
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
|
||||
CPL Configuration
|
||||
db_context: Optional[:class:`cpl.core.database.context.database_context_abc.DatabaseContextABC`]
|
||||
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
|
||||
self._scope: Optional[ScopeABC] = None
|
||||
|
||||
def _find_service(self, service_type: type) -> Optional[ServiceDescriptor]:
|
||||
@@ -94,17 +89,14 @@ class ServiceProvider(ServiceProviderABC):
|
||||
elif issubclass(parameter.annotation, ServiceProviderABC):
|
||||
params.append(self)
|
||||
|
||||
elif issubclass(parameter.annotation, ApplicationEnvironmentABC):
|
||||
params.append(self._configuration.environment)
|
||||
|
||||
elif issubclass(parameter.annotation, DatabaseContextABC):
|
||||
params.append(self._database_context)
|
||||
elif issubclass(parameter.annotation, Environment):
|
||||
params.append(Environment)
|
||||
|
||||
elif issubclass(parameter.annotation, ConfigurationModelABC):
|
||||
params.append(self._configuration.get_configuration(parameter.annotation))
|
||||
params.append(Configuration.get(parameter.annotation))
|
||||
|
||||
elif issubclass(parameter.annotation, ConfigurationABC):
|
||||
params.append(self._configuration)
|
||||
elif issubclass(parameter.annotation, Configuration):
|
||||
params.append(Configuration)
|
||||
|
||||
else:
|
||||
params.append(self._get_service(parameter, origin_service_type))
|
||||
@@ -115,7 +107,6 @@ class ServiceProvider(ServiceProviderABC):
|
||||
if origin_service_type is None:
|
||||
origin_service_type = service_type
|
||||
|
||||
|
||||
for descriptor in self._service_descriptors:
|
||||
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
|
||||
if descriptor.implementation is not None:
|
||||
@@ -142,7 +133,7 @@ class ServiceProvider(ServiceProviderABC):
|
||||
else:
|
||||
descriptors.append(copy.deepcopy(descriptor))
|
||||
|
||||
sb = ScopeBuilder(ServiceProvider(descriptors, self._configuration, self._database_context))
|
||||
sb = ScopeBuilder(ServiceProvider(descriptors, self._database_context))
|
||||
return sb.build()
|
||||
|
||||
def get_service(self, service_type: T, *args, **kwargs) -> Optional[R]:
|
||||
@@ -3,12 +3,12 @@ from abc import abstractmethod, ABC
|
||||
from inspect import Signature, signature
|
||||
from typing import Optional
|
||||
|
||||
from cpl.core.dependency_injection.scope_abc import ScopeABC
|
||||
from cpl.dependency.scope_abc import ScopeABC
|
||||
from cpl.core.typing import T, R
|
||||
|
||||
|
||||
class ServiceProviderABC(ABC):
|
||||
r"""ABC for the class :class:`cpl.core.dependency_injection.service_provider.ServiceProvider`"""
|
||||
r"""ABC for the class :class:`cpl.dependency.service_provider.ServiceProvider`"""
|
||||
|
||||
_provider: Optional["ServiceProviderABC"] = None
|
||||
|
||||
@@ -44,7 +44,7 @@ class ServiceProviderABC(ABC):
|
||||
|
||||
Parameter
|
||||
---------
|
||||
Object of type :class:`cpl.core.dependency_injection.scope_abc.ScopeABC`
|
||||
Object of type :class:`cpl.dependency.scope_abc.ScopeABC`
|
||||
Service scope
|
||||
"""
|
||||
|
||||
@@ -54,7 +54,7 @@ class ServiceProviderABC(ABC):
|
||||
|
||||
Returns
|
||||
-------
|
||||
Object of type :class:`cpl.core.dependency_injection.scope_abc.ScopeABC`
|
||||
Object of type :class:`cpl.dependency.scope_abc.ScopeABC`
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
30
src/cpl-dependency/pyproject.toml
Normal file
30
src/cpl-dependency/pyproject.toml
Normal file
@@ -0,0 +1,30 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "cpl-dependency"
|
||||
version = "2024.7.0"
|
||||
description = "CPL dependency"
|
||||
readme ="CPL dependency package"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
authors = [
|
||||
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
|
||||
]
|
||||
keywords = ["cpl", "dependency", "backend", "shared", "library"]
|
||||
|
||||
dynamic = ["dependencies", "optional-dependencies"]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://www.sh-edraft.de"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
include = ["cpl*"]
|
||||
|
||||
[tool.setuptools.dynamic]
|
||||
dependencies = { file = ["requirements.txt"] }
|
||||
optional-dependencies.dev = { file = ["requirements.dev.txt"] }
|
||||
|
||||
|
||||
1
src/cpl-dependency/requirements.dev.txt
Normal file
1
src/cpl-dependency/requirements.dev.txt
Normal file
@@ -0,0 +1 @@
|
||||
black==25.1.0
|
||||
1
src/cpl-dependency/requirements.txt
Normal file
1
src/cpl-dependency/requirements.txt
Normal file
@@ -0,0 +1 @@
|
||||
cpl-core
|
||||
@@ -1,3 +1,4 @@
|
||||
from cpl.dependency import ServiceCollection as _ServiceCollection
|
||||
from .abc.email_client_abc import EMailClientABC
|
||||
from .email_client import EMailClient
|
||||
from .email_client_settings import EMailClientSettings
|
||||
@@ -6,21 +7,15 @@ from .email_model import EMail
|
||||
from .mail_logger import MailLogger
|
||||
|
||||
|
||||
def add_mail(self):
|
||||
def add_mail(collection: _ServiceCollection):
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.log import LoggerABC
|
||||
|
||||
try:
|
||||
self.add_singleton(EMailClientABC, EMailClient)
|
||||
self.add_transient(LoggerABC, MailLogger)
|
||||
collection.add_singleton(EMailClientABC, EMailClient)
|
||||
collection.add_transient(LoggerABC, MailLogger)
|
||||
except ImportError as e:
|
||||
Console.error("cpl-translation is not installed", str(e))
|
||||
|
||||
|
||||
def init():
|
||||
from cpl.core.dependency_injection import ServiceCollection
|
||||
|
||||
ServiceCollection.add_mail = add_mail
|
||||
|
||||
|
||||
init()
|
||||
_ServiceCollection.with_module(add_mail, __name__)
|
||||
|
||||
@@ -2,7 +2,6 @@ import ssl
|
||||
from smtplib import SMTP
|
||||
from typing import Optional
|
||||
|
||||
from cpl.core.environment.application_environment_abc import ApplicationEnvironmentABC
|
||||
from cpl.core.utils.credential_manager import CredentialManager
|
||||
from cpl.mail.abc.email_client_abc import EMailClientABC
|
||||
from cpl.mail.email_client_settings import EMailClientSettings
|
||||
@@ -22,10 +21,11 @@ class EMailClient(EMailClientABC):
|
||||
Settings for mailing
|
||||
"""
|
||||
|
||||
def __init__(self, environment: ApplicationEnvironmentABC, logger: MailLogger, mail_settings: EMailClientSettings):
|
||||
def __init__(self, logger: MailLogger, mail_settings: EMailClientSettings):
|
||||
EMailClientABC.__init__(self)
|
||||
|
||||
self._environment = environment
|
||||
assert mail_settings is not None, "mail_settings must not be None"
|
||||
|
||||
self._mail_settings = mail_settings
|
||||
self._logger = logger
|
||||
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
from cpl.dependency import ServiceCollection as _ServiceCollection
|
||||
from .translate_pipe import TranslatePipe
|
||||
from .translation_service import TranslationService
|
||||
from .translation_service_abc import TranslationServiceABC
|
||||
from .translation_settings import TranslationSettings
|
||||
|
||||
|
||||
def add_translation(self):
|
||||
def add_translation(collection: _ServiceCollection):
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.pipes import PipeABC
|
||||
from cpl.translation.translate_pipe import TranslatePipe
|
||||
@@ -12,16 +13,10 @@ def add_translation(self):
|
||||
from cpl.translation.translation_service_abc import TranslationServiceABC
|
||||
|
||||
try:
|
||||
self.add_singleton(TranslationServiceABC, TranslationService)
|
||||
self.add_transient(PipeABC, TranslatePipe)
|
||||
collection.add_singleton(TranslationServiceABC, TranslationService)
|
||||
collection.add_transient(PipeABC, TranslatePipe)
|
||||
except ImportError as e:
|
||||
Console.error("cpl-translation is not installed", str(e))
|
||||
|
||||
|
||||
def init():
|
||||
from cpl.core.dependency_injection import ServiceCollection
|
||||
|
||||
ServiceCollection.add_translation = add_translation
|
||||
|
||||
|
||||
init()
|
||||
_ServiceCollection.with_module(add_translation, __name__)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from cpl.core.application import ApplicationABC
|
||||
from cpl.application import ApplicationABC
|
||||
from cpl.core.configuration import ConfigurationABC
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.dependency_injection import ServiceProviderABC
|
||||
from cpl.dependency import ServiceProviderABC
|
||||
|
||||
|
||||
class Application(ApplicationABC):
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import asyncio
|
||||
from cpl.core.application import ApplicationBuilder
|
||||
from cpl.application import ApplicationBuilder
|
||||
|
||||
from application import Application
|
||||
from startup import Startup
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from cpl.core.application.async_startup_abc import AsyncStartupABC
|
||||
from cpl.application.async_startup_abc import AsyncStartupABC
|
||||
from cpl.core.configuration import ConfigurationABC
|
||||
from cpl.core.dependency_injection import ServiceProviderABC, ServiceCollectionABC
|
||||
from cpl.core.environment import ApplicationEnvironment
|
||||
from cpl.dependency import ServiceProviderABC, ServiceCollection
|
||||
from cpl.core.environment import Environment
|
||||
|
||||
|
||||
class Startup(AsyncStartupABC):
|
||||
@@ -9,11 +9,9 @@ class Startup(AsyncStartupABC):
|
||||
AsyncStartupABC.__init__(self)
|
||||
|
||||
async def configure_configuration(
|
||||
self, configuration: ConfigurationABC, environment: ApplicationEnvironment
|
||||
self, configuration: ConfigurationABC, environment: Environment
|
||||
) -> ConfigurationABC:
|
||||
return configuration
|
||||
|
||||
async def configure_services(
|
||||
self, services: ServiceCollectionABC, environment: ApplicationEnvironment
|
||||
) -> ServiceProviderABC:
|
||||
async def configure_services(self, services: ServiceCollection, environment: Environment) -> ServiceProviderABC:
|
||||
return services.build_service_provider()
|
||||
|
||||
@@ -25,6 +25,7 @@ if __name__ == "__main__":
|
||||
Console.spinner(
|
||||
"Test:", test_spinner, spinner_foreground_color=ForegroundColorEnum.cyan, text_foreground_color="green"
|
||||
)
|
||||
Console.write_line("HOLD BACK")
|
||||
# opts = [
|
||||
# 'Option 1',
|
||||
# 'Option 2',
|
||||
|
||||
@@ -1,17 +1,17 @@
|
||||
from typing import Optional
|
||||
|
||||
from cpl.core.application import ApplicationABC
|
||||
from cpl.core.configuration import ConfigurationABC
|
||||
from cpl.application import ApplicationABC
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.dependency_injection import ServiceProviderABC
|
||||
from cpl.core.environment import Environment
|
||||
from cpl.core.log import LoggerABC
|
||||
from model.user_repo_abc import UserRepoABC
|
||||
from cpl.dependency import ServiceProviderABC
|
||||
from model.user_repo import UserRepo
|
||||
from model.user_repo_abc import UserRepoABC
|
||||
|
||||
|
||||
class Application(ApplicationABC):
|
||||
def __init__(self, config: ConfigurationABC, services: ServiceProviderABC):
|
||||
ApplicationABC.__init__(self, config, services)
|
||||
def __init__(self, services: ServiceProviderABC):
|
||||
ApplicationABC.__init__(self, services)
|
||||
|
||||
self._logger: Optional[LoggerABC] = None
|
||||
|
||||
@@ -19,10 +19,8 @@ class Application(ApplicationABC):
|
||||
self._logger = self._services.get_service(LoggerABC)
|
||||
|
||||
def main(self):
|
||||
self._logger.header(f"{self._configuration.environment.application_name}:")
|
||||
self._logger.debug(f"Host: {self._configuration.environment.host_name}")
|
||||
self._logger.debug(f"Environment: {self._configuration.environment.environment_name}")
|
||||
self._logger.debug(f"Customer: {self._configuration.environment.customer}")
|
||||
self._logger.debug(f"Host: {Environment.get_host_name()}")
|
||||
self._logger.debug(f"Environment: {Environment.get_environment()}")
|
||||
|
||||
user_repo: UserRepo = self._services.get_service(UserRepoABC)
|
||||
if len(user_repo.get_users()) == 0:
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from cpl.core.application import ApplicationBuilder
|
||||
from cpl.application import ApplicationBuilder
|
||||
|
||||
from application import Application
|
||||
from startup import Startup
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from cpl.core.database import TableABC
|
||||
from cpl.database import TableABC
|
||||
|
||||
|
||||
class CityModel(TableABC):
|
||||
|
||||
@@ -1,7 +0,0 @@
|
||||
from cpl.core.database import DatabaseSettings
|
||||
from cpl.core.database.context import DatabaseContext
|
||||
|
||||
|
||||
class DBContext(DatabaseContext):
|
||||
def __init__(self):
|
||||
DatabaseContext.__init__(self)
|
||||
@@ -1,4 +1,4 @@
|
||||
from cpl.core.database import TableABC
|
||||
from cpl.database import TableABC
|
||||
|
||||
from .city_model import CityModel
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.database.context import DatabaseContextABC
|
||||
from cpl.database.mysql.context import DatabaseContextABC
|
||||
|
||||
from .city_model import CityModel
|
||||
from .user_model import UserModel
|
||||
|
||||
@@ -1,11 +1,9 @@
|
||||
from cpl.core.application import StartupABC
|
||||
from cpl.core.configuration import ConfigurationABC
|
||||
from cpl.core.database import DatabaseSettings
|
||||
from cpl.core.dependency_injection import ServiceCollectionABC, ServiceProviderABC
|
||||
from cpl.core.environment import ApplicationEnvironmentABC
|
||||
from cpl.application import StartupABC
|
||||
from cpl.core.configuration import Configuration
|
||||
from cpl.core.environment import Environment
|
||||
from cpl.core.log import Logger, LoggerABC
|
||||
|
||||
from model.db_context import DBContext
|
||||
from cpl.database import mysql
|
||||
from cpl.dependency import ServiceCollection
|
||||
from model.user_repo import UserRepo
|
||||
from model.user_repo_abc import UserRepoABC
|
||||
|
||||
@@ -16,28 +14,15 @@ class Startup(StartupABC):
|
||||
|
||||
self._configuration = None
|
||||
|
||||
def configure_configuration(
|
||||
self, configuration: ConfigurationABC, environment: ApplicationEnvironmentABC
|
||||
) -> ConfigurationABC:
|
||||
configuration.add_environment_variables("PYTHON_")
|
||||
configuration.add_environment_variables("CPL_")
|
||||
def configure_configuration(self, configuration: Configuration, environment: Environment):
|
||||
configuration.add_json_file(f"appsettings.json")
|
||||
configuration.add_json_file(f"appsettings.{configuration.environment.environment_name}.json")
|
||||
configuration.add_json_file(f"appsettings.{configuration.environment.host_name}.json", optional=True)
|
||||
configuration.add_json_file(f"appsettings.{environment.get_environment()}.json")
|
||||
configuration.add_json_file(f"appsettings.{environment.get_host_name()}.json", optional=True)
|
||||
|
||||
self._configuration = configuration
|
||||
|
||||
return configuration
|
||||
|
||||
def configure_services(
|
||||
self, services: ServiceCollectionABC, environment: ApplicationEnvironmentABC
|
||||
) -> ServiceProviderABC:
|
||||
# Create and connect to database
|
||||
self._configuration.parse_console_arguments(services)
|
||||
db_settings: DatabaseSettings = self._configuration.get_configuration(DatabaseSettings)
|
||||
services.add_db_context(DBContext, db_settings)
|
||||
def configure_services(self, services: ServiceCollection, environment: Environment):
|
||||
services.add_module(mysql)
|
||||
|
||||
services.add_singleton(UserRepoABC, UserRepo)
|
||||
|
||||
services.add_singleton(LoggerABC, Logger)
|
||||
return services.build_service_provider()
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
from cpl.core.application import ApplicationABC
|
||||
from cpl.application import ApplicationABC
|
||||
from cpl.core.configuration import ConfigurationABC
|
||||
from cpl.core.console.console import Console
|
||||
from cpl.core.dependency_injection import ServiceProviderABC
|
||||
from cpl.core.dependency_injection.scope import Scope
|
||||
from cpl.dependency import ServiceProviderABC
|
||||
from cpl.dependency.scope import Scope
|
||||
from di.static_test import StaticTest
|
||||
from di.test_abc import TestABC
|
||||
from di.test_service import TestService
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from cpl.core.application import ApplicationBuilder
|
||||
from cpl.application import ApplicationBuilder
|
||||
|
||||
from di.application import Application
|
||||
from di.startup import Startup
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from cpl.core.application import StartupABC
|
||||
from cpl.application import StartupABC
|
||||
from cpl.core.configuration import ConfigurationABC
|
||||
from cpl.core.dependency_injection import ServiceProviderABC, ServiceCollectionABC
|
||||
from cpl.core.environment import ApplicationEnvironment
|
||||
from cpl.dependency import ServiceProviderABC, ServiceCollection
|
||||
from cpl.core.environment import Environment
|
||||
from di.test1_service import Test1Service
|
||||
from di.test2_service import Test2Service
|
||||
from di.test_abc import TestABC
|
||||
@@ -14,14 +14,10 @@ class Startup(StartupABC):
|
||||
def __init__(self):
|
||||
StartupABC.__init__(self)
|
||||
|
||||
def configure_configuration(
|
||||
self, configuration: ConfigurationABC, environment: ApplicationEnvironment
|
||||
) -> ConfigurationABC:
|
||||
def configure_configuration(self, configuration: ConfigurationABC, environment: Environment) -> ConfigurationABC:
|
||||
return configuration
|
||||
|
||||
def configure_services(
|
||||
self, services: ServiceCollectionABC, environment: ApplicationEnvironment
|
||||
) -> ServiceProviderABC:
|
||||
def configure_services(self, services: ServiceCollection, environment: Environment) -> ServiceProviderABC:
|
||||
services.add_scoped(TestService)
|
||||
services.add_scoped(DITesterService)
|
||||
|
||||
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user