2022.12 #133
@ -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
|
||||
|
@ -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
|
||||
|
@ -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]))
|
||||
|
@ -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:
|
||||
|
@ -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()
|
||||
|
@ -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
13
tests/custom/di/src/di/test1_service.py
Normal 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
13
tests/custom/di/src/di/test2_service.py
Normal 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
10
tests/custom/di/src/di/test_abc.py
Normal 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}>'
|
@ -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
9
tests/custom/di/src/di/tester.py
Normal 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)
|
Loading…
Reference in New Issue
Block a user