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

Improved service provider

Browse Source
This commit is contained in:
Sven Heidemann 2021-03-23 21:39:29 +01:00
parent a683b70c5f
commit ca51f88d2b
15 changed files with 93 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/cpl/configuration/configuration.py
View File

@ -11,7 +11,7 @@ from cpl.configuration.console_argument import ConsoleArgument
from cpl.console.console import Console
from cpl.console.foreground_color_enum import ForegroundColorEnum
from cpl.environment.application_environment import ApplicationEnvironment
from cpl.environment.environment_abc import ApplicationEnvironmentABC
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.environment.environment_name_enum import EnvironmentNameEnum

2
src/cpl/configuration/configuration_abc.py
View File

@ -4,7 +4,7 @@ from typing import Type, Union, Optional
from cpl.configuration.console_argument import ConsoleArgument
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.environment.environment_abc import ApplicationEnvironmentABC
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
class ConfigurationABC(ABC):

11
src/cpl/dependency_injection/service_collection.py
View File

@ -3,8 +3,6 @@ from typing import Union, Type, Callable, Optional
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.database.context import DatabaseContextABC
from cpl.dependency_injection.service_factory import ServiceFactory
from cpl.dependency_injection.service_factory_abc import ServiceFactoryABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
@ -46,12 +44,12 @@ class ServiceCollection(ServiceCollectionABC):
def add_singleton(self, service_type: Union[type, object], service: Union[type, object] = None):
if service is not None:
if isinstance(service, type):
service = service()
service = self.build_service_provider().build_service(service)
self._add_descriptor(service, ServiceLifetimeEnum.singleton)
else:
if isinstance(service_type, type):
service_type = service_type()
service_type = self.build_service_provider().build_service(service_type)
self._add_descriptor(service_type, ServiceLifetimeEnum.singleton)
@ -64,8 +62,5 @@ class ServiceCollection(ServiceCollectionABC):
else:
self._add_descriptor(service_type, ServiceLifetimeEnum.transient)
def build_service_factory(self) -> ServiceFactoryABC:
return ServiceFactory(self._service_descriptors, self._configuration, self._runtime)
def build_service_provider(self) -> ServiceProviderABC:
return ServiceProvider(self.build_service_factory())
return ServiceProvider(self._service_descriptors, self._configuration, self._runtime)

27
src/cpl/dependency_injection/service_factory.py
View File

@ -1,27 +0,0 @@
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.dependency_injection.service_factory_abc import ServiceFactoryABC
class ServiceFactory(ServiceFactoryABC):
def __init__(self, service_descriptors: list[ServiceDescriptor], config: ConfigurationABC,
runtime: ApplicationRuntimeABC):
ServiceFactoryABC.__init__(self)
self._service_descriptors: list[ServiceDescriptor] = service_descriptors
self._configuration: ConfigurationABC = config
self._runtime: ApplicationRuntimeABC = runtime
@property
def service_descriptors(self) -> list[ServiceDescriptor]:
return self._service_descriptors
@property
def configuration(self) -> ConfigurationABC:
return self._configuration
@property
def runtime(self) -> ApplicationRuntimeABC:
return self._runtime

23
src/cpl/dependency_injection/service_factory_abc.py
View File

@ -1,23 +0,0 @@
from abc import ABC, abstractmethod
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
class ServiceFactoryABC(ABC):
@abstractmethod
def __init__(self): pass
@property
@abstractmethod
def service_descriptors(self) -> list[ServiceDescriptor]: pass
@property
@abstractmethod
def configuration(self) -> ConfigurationABC: pass
@property
@abstractmethod
def runtime(self) -> ApplicationRuntimeABC: pass

68
src/cpl/dependency_injection/service_provider.py
View File

@ -1,26 +1,82 @@
from collections import Callable
from inspect import signature, Parameter
from typing import Optional
from cpl.dependency_injection import ServiceProviderABC
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.dependency_injection.service_factory_abc import ServiceFactoryABC
from cpl.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
class ServiceProvider(ServiceProviderABC):
def __init__(self, service_factory: ServiceFactoryABC):
def __init__(self, service_descriptors: list[ServiceDescriptor], config: ConfigurationABC,
runtime: ApplicationRuntimeABC):
ServiceProviderABC.__init__(self)
self._service_factory = service_factory
self._service_descriptors: list[ServiceDescriptor] = service_descriptors
self._configuration: ConfigurationABC = config
self._runtime: ApplicationRuntimeABC = runtime
def _find_service(self, service_type: type) -> [ServiceDescriptor]:
for descriptor in self._service_factory.service_descriptors:
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
return descriptor
return None
def _get_service(self, service_type: type, parameter: Parameter) -> object:
for descriptor in self._service_descriptors:
if descriptor.service_type == parameter.annotation or issubclass(descriptor.service_type,
parameter.annotation):
if descriptor.implementation is not None:
return descriptor.implementation
implementation = self.build_service(service_type)
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptor.implementation = implementation
return descriptor.implementation
def build_service(self, service_type: type) -> object:
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
if descriptor.implementation is not None:
service_type = type(descriptor.implementation)
else:
service_type = descriptor.service_type
sig = signature(service_type.__init__)
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
elif issubclass(parameter.annotation, ApplicationRuntimeABC):
params.append(self._runtime)
elif issubclass(parameter.annotation, ApplicationEnvironmentABC):
params.append(self._configuration.environment)
# elif issubclass(parameter.annotation, DatabaseContextABC):
# params.append(self._database_context)
elif issubclass(parameter.annotation, ConfigurationModelABC):
params.append(self._configuration.get_configuration(parameter.annotation))
elif issubclass(parameter.annotation, ConfigurationABC):
params.append(self._configuration)
else:
params.append(self._get_service(service_type, parameter))
return service_type(*params)
def get_service(self, service_type: type) -> Optional[Callable[object]]:
result = self._find_service(service_type)
@ -30,7 +86,7 @@ class ServiceProvider(ServiceProviderABC):
if result.implementation is not None:
return result.implementation
implementation = result.service_type()
implementation = self.build_service(service_type)
if result.lifetime == ServiceLifetimeEnum.singleton:
result.implementation = implementation

3
src/cpl/dependency_injection/service_provider_abc.py
View File

@ -14,6 +14,9 @@ class ServiceProviderABC(ABC):
"""
pass
@abstractmethod
def build_service(self, service_type: type) -> object: pass
@abstractmethod
def get_service(self, instance_type: Type) -> Callable[ServiceABC]:
"""

122
src/cpl/dependency_injection/service_provider_old.py
View File

@ -1,122 +0,0 @@
from collections import Callable
from inspect import signature, Parameter
from typing import Type, Optional, Union
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_abc import ServiceABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.environment.environment_abc import ApplicationEnvironmentABC
class ServiceProvider(ServiceProviderABC):
def __init__(self, config: ConfigurationABC, runtime: ApplicationRuntimeABC):
"""
Service for service providing
:param runtime:
"""
ServiceProviderABC.__init__(self)
self._configuration: ConfigurationABC = config
self._runtime: ApplicationRuntimeABC = runtime
self._database_context: Optional[DatabaseContextABC] = None
self._transient_services: dict[Type[ServiceABC], Callable[ServiceABC]] = {}
self._scoped_services: dict[Type[ServiceABC], Callable[ServiceABC]] = {}
self._singleton_services: dict[Type[ServiceABC], Callable[ServiceABC], ServiceABC] = {}
def _create_instance(self, service: Union[Callable[ServiceABC], ServiceABC]) -> Callable[ServiceABC]:
"""
Creates an instance of given type
:param service:
:return:
"""
sig = signature(service.__init__)
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if issubclass(parameter.annotation, ApplicationRuntimeABC):
params.append(self._runtime)
elif issubclass(parameter.annotation, ApplicationEnvironmentABC):
params.append(self._configuration.environment)
elif issubclass(parameter.annotation, DatabaseContextABC):
params.append(self._database_context)
elif issubclass(parameter.annotation, ConfigurationModelABC):
params.append(self._configuration.get_configuration(parameter.annotation))
elif issubclass(parameter.annotation, ConfigurationABC):
params.append(self._configuration)
elif issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
else:
params.append(self.get_service(parameter.annotation))
return service(*params)
def add_db_context(self, db_context: Type[DatabaseContextABC]):
self._database_context = self._create_instance(db_context)
def get_db_context(self) -> Callable[DatabaseContextABC]:
return self._database_context
def add_transient(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
if service is None:
self._transient_services[service_type] = service_type
else:
self._transient_services[service_type] = service
def add_scoped(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
if service is None:
self._scoped_services[service_type] = service_type
else:
self._scoped_services[service_type] = service
def add_singleton(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
for known_service in self._singleton_services:
if type(known_service) == service_type:
raise Exception(f'Service with type {service_type} already exists')
if service is None:
self._singleton_services[service_type] = self._create_instance(service_type)
else:
self._singleton_services[service_type] = self._create_instance(service)
def get_service(self, instance_type: Type) -> Callable[ServiceABC]:
if issubclass(instance_type, ServiceProviderABC):
return self
for service in self._transient_services:
if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
return self._create_instance(self._transient_services[service])
for service in self._scoped_services:
if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
return self._create_instance(self._scoped_services[service])
for service in self._singleton_services:
if service == instance_type and isinstance(self._singleton_services[service], instance_type):
return self._singleton_services[service]
def remove_service(self, instance_type: Type[ServiceABC]):
for service in self._transient_services:
if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
del self._transient_services[service]
return
for service in self._scoped_services:
if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
del self._scoped_services[service]
return
for service in self._singleton_services:
if service == instance_type and isinstance(self._singleton_services[service], instance_type):
del self._singleton_services[service]
return

2
src/cpl/environment/__init__.py
View File

@ -20,7 +20,7 @@ __version__ = '2021.4.2'
from collections import namedtuple
# imports:
from .environment_abc import ApplicationEnvironmentABC
from .application_environment_abc import ApplicationEnvironmentABC
from .environment_name_enum import EnvironmentNameEnum
from .application_environment import ApplicationEnvironment

2
src/cpl/environment/application_environment.py
View File

@ -1,7 +1,7 @@
from socket import gethostname
from typing import Optional
from cpl.environment.environment_abc import ApplicationEnvironmentABC
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.environment.environment_name_enum import EnvironmentNameEnum

0
src/cpl/environment/environment_abc.py → src/cpl/environment/application_environment_abc.py
View File

2
src/cpl/mailing/email_client_service.py
View File

@ -2,7 +2,7 @@ import ssl
from smtplib import SMTP
from typing import Optional
from cpl.environment.environment_abc import ApplicationEnvironmentABC
from cpl.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl.logging.logger_abc import LoggerABC
from cpl.mailing.email import EMail
from cpl.mailing.email_client_abc import EMailClientABC

5
src/tests/custom/general/application.py
View File

@ -9,6 +9,7 @@ from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.logging.logger_abc import LoggerABC
from cpl.mailing.email import EMail
from cpl.mailing.email_client_abc import EMailClientABC
from tests.custom.general.test_service import TestService
class Application(ApplicationABC):
@ -50,6 +51,8 @@ class Application(ApplicationABC):
self._logger.debug(__name__, f'Host: {self._configuration.environment.host_name}')
self._logger.debug(__name__, f'Environment: {self._configuration.environment.environment_name}')
self._logger.debug(__name__, f'Customer: {self._configuration.environment.customer}')
Console.spinner('Test', self._wait, 20, spinner_foreground_color='red')
Console.spinner('Test', self._wait, 2, spinner_foreground_color='red')
test: TestService = self._services.get_service(TestService)
test.run()
# self.test_send_mail()
# self.test_console()

3
src/tests/custom/general/startup.py
View File

@ -10,6 +10,7 @@ from cpl.logging.logger_abc import LoggerABC
from cpl.mailing.email_client_service import EMailClient
from cpl.mailing.email_client_abc import EMailClientABC
from cpl.utils.credential_manager import CredentialManager
from tests.custom.general.test_service import TestService
class Startup(StartupABC):
@ -20,7 +21,6 @@ class Startup(StartupABC):
self._configuration = config
self._application_runtime = runtime
self._services = services
print(self._services)
def configure_configuration(self) -> ConfigurationABC:
self._configuration.add_environment_variables('PYTHON_')
@ -40,5 +40,6 @@ class Startup(StartupABC):
self._services.add_singleton(LoggerABC, Logger)
self._services.add_singleton(EMailClientABC, EMailClient)
self._services.add_singleton(TestService)
return self._services.build_service_provider()

13
src/tests/custom/general/test_service.py Normal file
View File

@ -0,0 +1,13 @@
from cpl.console.console import Console
from cpl.dependency_injection import ServiceABC, ServiceProviderABC
class TestService(ServiceABC):
def __init__(self, provider: ServiceProviderABC):
ServiceABC.__init__(self)
self._provider = provider
def run(self):
Console.write_line('Hello World!', self._provider)