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

Merge pull request 'Added multiple instance handling (#152)' (#156) from #152 into 2022.12 

Reviewed-on: #156
Closes #152
This commit is contained in:
Sven Heidemann 2022-12-23 17:29:15 +01:00
parent a507ed9f46 59263ece6e
commit 7af83b49a5
11 changed files with 146 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,17 +1,16 @@
import copy
import functools
import typing
from inspect import signature, Parameter, Signature
from typing import Optional, Callable
from typing import Optional
from cpl_core.configuration.configuration_abc import ConfigurationABC
from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl_core.console import Console
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_provider_abc import ServiceProviderABC
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.type import T
@ -44,7 +43,7 @@ class ServiceProvider(ServiceProviderABC):
return None
def _get_service(self, parameter: Parameter) -> object:
def _get_service(self, parameter: Parameter) -> Optional[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:
@ -58,12 +57,32 @@ class ServiceProvider(ServiceProviderABC):
# raise Exception(f'Service {parameter.annotation} not found')
def _get_services(self, t: type) -> list[Optional[object]]:
implementations = []
for descriptor in self._service_descriptors:
if descriptor.service_type == t or issubclass(descriptor.service_type, t):
if descriptor.implementation is not None:
implementations.append(descriptor.implementation)
continue
implementation = self.build_service(descriptor.service_type)
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptor.implementation = implementation
implementations.append(implementation)
return implementations
def build_by_signature(self, sig: Signature) -> list[T]:
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if issubclass(parameter.annotation, ServiceProviderABC):
if typing.get_origin(parameter.annotation) == list:
params.append(self._get_services(typing.get_args(parameter.annotation)[0]))
elif issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
elif issubclass(parameter.annotation, ApplicationEnvironmentABC):
@ -83,7 +102,7 @@ class ServiceProvider(ServiceProviderABC):
return params
def build_service(self, service_type: T) -> object:
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:
@ -120,17 +139,12 @@ class ServiceProvider(ServiceProviderABC):
return implementation
@classmethod
def inject(cls, f=None):
if f is None:
return functools.partial(cls.inject)
def get_services(self, service_type: T) -> list[Optional[T]]:
implementations = []
@functools.wraps(f)
def inner(*args, **kwargs):
if cls._provider is None:
raise Exception(f'{cls.__name__} not build!')
if typing.get_origin(service_type) != list:
raise Exception(f'Invalid type {service_type}! Expected list of type')
injection = cls._provider.build_by_signature(signature(f))
return f(*injection, *args, **kwargs)
implementations.extend(self._get_services(typing.get_args(service_type)[0]))
return inner
return implementations

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

@ -1,8 +1,8 @@
import functools
from abc import abstractmethod, ABC
from inspect import Signature
from inspect import Signature, signature
from typing import Type, Optional
from cpl_core.dependency_injection.scope_abc import ScopeABC
from cpl_core.type import T
@ -23,12 +23,12 @@ class ServiceProviderABC(ABC):
def build_by_signature(self, sig: Signature) -> list[T]: pass
@abstractmethod
def build_service(self, service_type: T) -> object:
def build_service(self, service_type: type) -> object:
r"""Creates instance of given type
Parameter
---------
instance_type: :class:`Type`
instance_type: :class:`type`
The type of the searched instance
Returns
@ -36,25 +36,25 @@ class ServiceProviderABC(ABC):
Object of the given type
"""
pass
@abstractmethod
def set_scope(self, scope: ScopeABC):
r"""Sets the scope of service provider
Parameter
---------
scope :class:`cpl_core.dependency_injection.scope.Scope`
Object of type :class:`cpl_core.dependency_injection.scope_abc.ScopeABC`
Service scope
"""
pass
@abstractmethod
def create_scope(self) -> ScopeABC:
r"""Creates a service scope
Returns
-------
Object of type :class:`cpl_core.dependency_injection.scope.Scope`
Object of type :class:`cpl_core.dependency_injection.scope_abc.ScopeABC`
"""
pass
@ -64,16 +64,51 @@ class ServiceProviderABC(ABC):
Parameter
---------
instance_type: :class:`Type`
instance_type: :class:`cpl_core.type.T`
The type of the searched instance
Returns
-------
Object of type Optional[Callable[:class:`object`]]
Object of type Optional[:class:`cpl_core.type.T`]
"""
pass
# @classmethod
# @abstractmethod
# def inject(cls):
# pass
@abstractmethod
def get_services(self, service_type: T) -> list[Optional[T]]:
r"""Returns instance of given type
Parameter
---------
instance_type: :class:`cpl_core.type.T`
The type of the searched instance
Returns
-------
Object of type list[Optional[:class:`cpl_core.type.T`]
"""
pass
@classmethod
def inject(cls, f=None):
r"""Decorator to allow injection into static and class methods
Parameter
---------
f: Callable
Returns
-------
function
"""
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

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

@ -4,8 +4,10 @@ from cpl_core.console.console import Console
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.dependency_injection.scope import Scope
from di.static_test import StaticTest
from di.test_service_service import TestService
from di.test_abc import TestABC
from di.test_service import TestService
from di.di_tester_service import DITesterService
from di.tester import Tester
class Application(ApplicationABC):
@ -28,10 +30,6 @@ class Application(ApplicationABC):
dit: DITesterService = scope.service_provider.get_service(DITesterService)
dit.run()
# Console.write_line('Disposed:')
# ts1: TestService = scope1.service_provider.get_service(TestService)
# ts1.run()
with self._services.create_scope() as scope:
Console.write_line('Scope2')
ts: TestService = scope.service_provider.get_service(TestService)
@ -42,5 +40,6 @@ class Application(ApplicationABC):
Console.write_line('Global')
self._part_of_scoped()
StaticTest.test()
with self._services.create_scope() as scope:
StaticTest.test()
self._services.get_service(Tester)
Console.write_line(self._services.get_services(list[TestABC]))

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

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

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

@ -2,8 +2,12 @@ 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 di.test_service_service import TestService
from di.test1_service import Test1Service
from di.test2_service import Test2Service
from di.test_abc import TestABC
from di.test_service import TestService
from di.di_tester_service import DITesterService
from di.tester import Tester
class Startup(StartupABC):
@ -17,5 +21,9 @@ class Startup(StartupABC):
def configure_services(self, services: ServiceCollectionABC, environment: ApplicationEnvironment) -> ServiceProviderABC:
services.add_scoped(TestService)
services.add_scoped(DITesterService)
services.add_singleton(TestABC, Test1Service)
services.add_singleton(TestABC, Test2Service)
services.add_singleton(Tester)
return services.build_service_provider()

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

@ -1,6 +1,6 @@
from cpl_core.configuration import ConfigurationABC
from cpl_core.dependency_injection import ServiceProvider, ServiceProviderABC
from di.test_service_service import TestService
from di.test_service import TestService
class StaticTest:

13
tests/custom/di/src/di/test1_service.py Normal file
View File

@ -0,0 +1,13 @@
import string
from cpl_core.console.console import Console
from cpl_core.utils.string import String
from di.test_abc import TestABC
class Test1Service(TestABC):
def __init__(self):
TestABC.__init__(self, String.random_string(string.ascii_lowercase, 8))
def run(self):
Console.write_line(f'Im {self._name}')

13
tests/custom/di/src/di/test2_service.py Normal file
View File

@ -0,0 +1,13 @@
import string
from cpl_core.console.console import Console
from cpl_core.utils.string import String
from di.test_abc import TestABC
class Test2Service(TestABC):
def __init__(self):
TestABC.__init__(self, String.random_string(string.ascii_lowercase, 8))
def run(self):
Console.write_line(f'Im {self._name}')

10
tests/custom/di/src/di/test_abc.py Normal file
View File

@ -0,0 +1,10 @@
from abc import ABC
class TestABC(ABC):
def __init__(self, name: str):
self._name = name
def __repr__(self):
return f'<{type(self).__name__} {self._name}>'

6
tests/custom/di/src/di/test_service_service.py → tests/custom/di/src/di/test_service.py
View File

@ -1,4 +1,5 @@
import string
from cpl_core.console.console import Console
from cpl_core.utils.string import String
@ -7,7 +8,6 @@ class TestService:
def __init__(self):
self._name = String.random_string(string.ascii_lowercase, 8)
def run(self):
Console.write_line(f'Im {self._name}')
Console.write_line(f'Im {self._name}')

9
tests/custom/di/src/di/tester.py Normal file
View File

@ -0,0 +1,9 @@
from cpl_core.console.console import Console
from di.test_abc import TestABC
class Tester:
def __init__(self, t1: TestABC, t2: TestABC, t3: list[TestABC]):
Console.write_line('Tester:')
Console.write_line(t1, t2, t3)