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

Merge pull request 'Static dependency injection (#148)' (#155) from #148 into 2022.12

Browse Source 
Reviewed-on: #155
Closes #148
This commit is contained in:
Sven Heidemann 2022-12-23 16:36:21 +01:00
commit 6b451142e2
11 changed files with 120 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -268,7 +268,7 @@ class Configuration(ConfigurationABC):
configuration.from_dict(value)
self.add_configuration(sub, configuration)
def add_configuration(self, key_type: Union[str, type], value: any):
def add_configuration(self, key_type: T, value: any):
self._config[key_type] = value
def create_console_argument(self, arg_type: ArgumentTypeEnum, token: str, name: str, aliases: list[str],

4
src/cpl_core/configuration/configuration_abc.py
View File

@ -76,7 +76,7 @@ class ConfigurationABC(ABC):
pass
@abstractmethod
def add_configuration(self, key_type: Union[str, type], value: any):
def add_configuration(self, key_type: T, value: any):
r"""Add configuration object
Parameter
@ -126,7 +126,7 @@ class ConfigurationABC(ABC):
pass
@abstractmethod
def get_configuration(self, search_type: Union[str, Type[ConfigurationModelABC]]) -> Optional[T]:
def get_configuration(self, search_type: Type[T]) -> Optional[T]:
r"""Returns value from configuration by given type
Parameter

13
src/cpl_core/dependency_injection/service_collection.py
View File

@ -11,6 +11,7 @@ from cpl_core.dependency_injection.service_provider_abc import ServiceProviderAB
from cpl_core.logging.logger_abc import LoggerABC
from cpl_core.logging.logger_service import Logger
from cpl_core.pipes.pipe_abc import PipeABC
from cpl_core.type import T
class ServiceCollection(ServiceCollectionABC):
@ -53,22 +54,26 @@ class ServiceCollection(ServiceCollectionABC):
def add_logging(self):
self.add_singleton(LoggerABC, Logger)
return self
def add_pipes(self):
for pipe in PipeABC.__subclasses__():
self.add_transient(PipeABC, pipe)
return self
def add_singleton(self, service_type: Union[type, object], service: Union[type, object] = None):
def add_singleton(self, service_type: T, service: T = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.singleton, service)
return self
def add_scoped(self, service_type: Type, service: Callable = None):
def add_scoped(self, service_type: T, service: Callable = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.scoped, service)
return self
def add_transient(self, service_type: type, service: type = None):
def add_transient(self, service_type: T, service: T = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.transient, service)
return self
def build_service_provider(self) -> ServiceProviderABC:
return ServiceProvider(self._service_descriptors, self._configuration, self._database_context)
sp = ServiceProvider(self._service_descriptors, self._configuration, self._database_context)
ServiceProviderABC.set_global_provider(sp)
return sp

7
src/cpl_core/dependency_injection/service_collection_abc.py
View File

@ -5,6 +5,7 @@ from typing import Type
from cpl_core.database.database_settings import DatabaseSettings
from cpl_core.database.context.database_context_abc import DatabaseContextABC
from cpl_core.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl_core.type import T
class ServiceCollectionABC(ABC):
@ -46,7 +47,7 @@ class ServiceCollectionABC(ABC):
pass
@abstractmethod
def add_transient(self, service_type: Type, service: Callable = None) -> 'ServiceCollectionABC':
def add_transient(self, service_type: T, service: T = None) -> 'ServiceCollectionABC':
r"""Adds a service with transient lifetime
Parameter
@ -63,7 +64,7 @@ class ServiceCollectionABC(ABC):
pass
@abstractmethod
def add_scoped(self, service_type: Type, service: Callable = None) -> 'ServiceCollectionABC':
def add_scoped(self, service_type: T, service: T = None) -> 'ServiceCollectionABC':
r"""Adds a service with scoped lifetime
Parameter
@ -80,7 +81,7 @@ class ServiceCollectionABC(ABC):
pass
@abstractmethod
def add_singleton(self, service_type: Type, service: Callable = None) -> 'ServiceCollectionABC':
def add_singleton(self, service_type: T, service: T = None) -> 'ServiceCollectionABC':
r"""Adds a service with singleton lifetime
Parameter

52
src/cpl_core/dependency_injection/service_provider.py
View File

@ -1,6 +1,7 @@
import copy
from inspect import signature, Parameter
from typing import Optional
import functools
from inspect import signature, Parameter, Signature
from typing import Optional, Callable
from cpl_core.configuration.configuration_abc import ConfigurationABC
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
@ -12,6 +13,7 @@ from cpl_core.dependency_injection.service_provider_abc import ServiceProviderAB
from cpl_core.dependency_injection.service_descriptor import ServiceDescriptor
from cpl_core.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl_core.environment.application_environment_abc import ApplicationEnvironmentABC
from cpl_core.type import T
class ServiceProvider(ServiceProviderABC):
@ -54,17 +56,9 @@ class ServiceProvider(ServiceProviderABC):
return implementation
def build_service(self, service_type: type) -> object:
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
if descriptor.implementation is not None:
service_type = type(descriptor.implementation)
else:
service_type = descriptor.service_type
# raise Exception(f'Service {parameter.annotation} not found')
break
sig = signature(service_type.__init__)
def build_by_signature(self, sig: Signature) -> list[T]:
params = []
for param in sig.parameters.items():
parameter = param[1]
@ -87,16 +81,31 @@ class ServiceProvider(ServiceProviderABC):
else:
params.append(self._get_service(parameter))
return params
def build_service(self, service_type: T) -> object:
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
if descriptor.implementation is not None:
service_type = type(descriptor.implementation)
else:
service_type = descriptor.service_type
break
sig = signature(service_type.__init__)
params = self.build_by_signature(sig)
return service_type(*params)
def set_scope(self, scope: ScopeABC):
self._scope = scope
def create_scope(self) -> ScopeABC:
sb = ScopeBuilder(ServiceProvider(self._service_descriptors, self._configuration, self._database_context))
sb = ScopeBuilder(ServiceProvider(copy.deepcopy(self._service_descriptors), self._configuration, self._database_context))
return sb.build()
def get_service(self, service_type: type) -> Optional[object]:
def get_service(self, service_type: T) -> Optional[T]:
result = self._find_service(service_type)
if result is None:
@ -110,3 +119,18 @@ class ServiceProvider(ServiceProviderABC):
result.implementation = implementation
return implementation
@classmethod
def inject(cls, f=None):
if f is None:
return functools.partial(cls.inject)
@functools.wraps(f)
def inner(*args, **kwargs):
if cls._provider is None:
raise Exception(f'{cls.__name__} not build!')
injection = cls._provider.build_by_signature(signature(f))
return f(*injection, *args, **kwargs)
return inner

26
src/cpl_core/dependency_injection/service_provider_abc.py
View File

@ -1,18 +1,29 @@
from abc import abstractmethod, ABC
from inspect import Signature
from typing import Type, Optional
from cpl_core.dependency_injection.scope_abc import ScopeABC
from cpl_core.type import T
class ServiceProviderABC(ABC):
r"""ABC for the class :class:`cpl_core.dependency_injection.service_provider.ServiceProvider`"""
@abstractmethod
def __init__(self):
pass
_provider: Optional['ServiceProviderABC'] = None
@abstractmethod
def build_service(self, service_type: Type) -> object:
def __init__(self): pass
@classmethod
def set_global_provider(cls, provider: 'ServiceProviderABC'):
cls._provider = provider
@abstractmethod
def build_by_signature(self, sig: Signature) -> list[T]: pass
@abstractmethod
def build_service(self, service_type: T) -> object:
r"""Creates instance of given type
Parameter
@ -48,7 +59,7 @@ class ServiceProviderABC(ABC):
pass
@abstractmethod
def get_service(self, instance_type: Type) -> Optional[object]:
def get_service(self, instance_type: T) -> Optional[T]:
r"""Returns instance of given type
Parameter
@ -61,3 +72,8 @@ class ServiceProviderABC(ABC):
Object of type Optional[Callable[:class:`object`]]
"""
pass
# @classmethod
# @abstractmethod
# def inject(cls):
# pass

41
tests/custom/di/src/di/application.py
View File

@ -3,8 +3,9 @@ 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 test_service_service import TestService
from di_tester_service import DITesterService
from di.static_test import StaticTest
from di.test_service_service import TestService
from di.di_tester_service import DITesterService
class Application(ApplicationABC):
@ -20,26 +21,26 @@ class Application(ApplicationABC):
pass
def main(self):
Console.write_line('Scope1')
scope1: Scope = self._services.create_scope()
ts: TestService = scope1.service_provider.get_service(TestService)
ts.run()
dit: DITesterService = scope1.service_provider.get_service(DITesterService)
dit.run()
t = scope1
b = t.service_provider
scope1.dispose()
with self._services.create_scope() as scope:
Console.write_line('Scope1')
ts: TestService = scope.service_provider.get_service(TestService)
ts.run()
dit: DITesterService = scope.service_provider.get_service(DITesterService)
dit.run()
#Console.write_line('Disposed:')
#ts1: TestService = scope1.service_provider.get_service(TestService)
#ts1.run()
# Console.write_line('Disposed:')
# ts1: TestService = scope1.service_provider.get_service(TestService)
# ts1.run()
Console.write_line('Scope2')
scope2: Scope = self._services.create_scope()
ts: TestService = scope2.service_provider.get_service(TestService)
ts.run()
dit: DITesterService = scope2.service_provider.get_service(DITesterService)
dit.run()
with self._services.create_scope() as scope:
Console.write_line('Scope2')
ts: TestService = scope.service_provider.get_service(TestService)
ts.run()
dit: DITesterService = scope.service_provider.get_service(DITesterService)
dit.run()
Console.write_line('Global')
self._part_of_scoped()
StaticTest.test()
with self._services.create_scope() as scope:
StaticTest.test()

5
tests/custom/di/src/di/di.json
View File

@ -16,7 +16,10 @@
"LicenseName": "",
"LicenseDescription": "",
"Dependencies": [
"sh_cpl>=2021.10.0.post1"
"cpl-core==2022.12.0"
],
"DevDependencies": [
"cpl-cli==2022.12.0"
],
"PythonVersion": ">=3.9.2",
"PythonPath": {},

2
tests/custom/di/src/di/di_tester_service.py
View File

@ -1,5 +1,5 @@
from cpl_core.console.console import Console
from test_service_service import TestService
from di.test_service_service import TestService
class DITesterService:

4
tests/custom/di/src/di/startup.py
View File

@ -2,8 +2,8 @@ from cpl_core.application import StartupABC
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceProviderABC, ServiceCollectionABC
from cpl_core.environment import ApplicationEnvironment
from test_service_service import TestService
from di_tester_service import DITesterService
from di.test_service_service import TestService
from di.di_tester_service import DITesterService
class Startup(StartupABC):

11
tests/custom/di/src/di/static_test.py Normal file
View File

@ -0,0 +1,11 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceProvider, ServiceProviderABC
from di.test_service_service import TestService
class StaticTest:
@staticmethod
@ServiceProvider.inject
def test(services: ServiceProviderABC, config: ConfigurationABC, t1: TestService):
t1.run()