2023.10 #175
@ -24,7 +24,7 @@ from cpl_core.dependency_injection.service_provider_abc import ServiceProviderAB
|
||||
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.type import T
|
||||
from cpl_core.type import T, R
|
||||
from cpl_core.utils.json_processor import JSONProcessor
|
||||
|
||||
|
||||
@ -316,7 +316,7 @@ class Configuration(ConfigurationABC):
|
||||
for arg in self._argument_types:
|
||||
call(arg)
|
||||
|
||||
def get_configuration(self, search_type: T) -> Optional[T]:
|
||||
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
|
||||
@ -357,7 +357,7 @@ class Configuration(ConfigurationABC):
|
||||
if exe.validators is not None:
|
||||
abort = False
|
||||
for validator_type in exe.validators:
|
||||
validator: ValidatorABC = services.get_service(validator_type)
|
||||
validator = services.get_service(validator_type)
|
||||
result = validator.validate()
|
||||
abort = not result
|
||||
if abort:
|
||||
@ -366,7 +366,7 @@ class Configuration(ConfigurationABC):
|
||||
if abort:
|
||||
sys.exit()
|
||||
|
||||
cmd: ArgumentExecutableABC = services.get_service(exe.executable_type)
|
||||
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")
|
||||
|
@ -6,7 +6,7 @@ from cpl_core.configuration.configuration_model_abc import ConfigurationModelABC
|
||||
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.type import T
|
||||
from cpl_core.type import T, R
|
||||
|
||||
|
||||
class ConfigurationABC(ABC):
|
||||
@ -124,7 +124,7 @@ class ConfigurationABC(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_configuration(self, search_type: T) -> Optional[T]:
|
||||
def get_configuration(self, search_type: T) -> Optional[R]:
|
||||
r"""Returns value from configuration by given type
|
||||
|
||||
Parameter:
|
||||
|
@ -61,15 +61,15 @@ class ServiceCollection(ServiceCollectionABC):
|
||||
self.add_transient(PipeABC, pipe)
|
||||
return self
|
||||
|
||||
def add_singleton(self, service_type: Type[T], service: T = 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[T], service: Callable = None):
|
||||
def add_scoped(self, service_type: T, service: T = None):
|
||||
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.scoped, service)
|
||||
return self
|
||||
|
||||
def add_transient(self, service_type: Type[T], service: T = None):
|
||||
def add_transient(self, service_type: T, service: T = None):
|
||||
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.transient, service)
|
||||
return self
|
||||
|
||||
|
@ -46,7 +46,7 @@ class ServiceCollectionABC(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def add_transient(self, service_type: Type[T], service: T = None) -> "ServiceCollectionABC":
|
||||
def add_transient(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
|
||||
r"""Adds a service with transient lifetime
|
||||
|
||||
Parameter:
|
||||
@ -61,7 +61,7 @@ class ServiceCollectionABC(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def add_scoped(self, service_type: Type[T], service: T = None) -> "ServiceCollectionABC":
|
||||
def add_scoped(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
|
||||
r"""Adds a service with scoped lifetime
|
||||
|
||||
Parameter:
|
||||
@ -76,7 +76,7 @@ class ServiceCollectionABC(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def add_singleton(self, service_type: Type[T], service: T = None) -> "ServiceCollectionABC":
|
||||
def add_singleton(self, service_type: T, service: T = None) -> "ServiceCollectionABC":
|
||||
r"""Adds a service with singleton lifetime
|
||||
|
||||
Parameter:
|
||||
|
@ -12,7 +12,7 @@ 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
|
||||
from cpl_core.type import T, R
|
||||
|
||||
|
||||
class ServiceProvider(ServiceProviderABC):
|
||||
@ -80,7 +80,7 @@ class ServiceProvider(ServiceProviderABC):
|
||||
|
||||
return implementations
|
||||
|
||||
def build_by_signature(self, sig: Signature) -> list[T]:
|
||||
def build_by_signature(self, sig: Signature) -> list[R]:
|
||||
params = []
|
||||
for param in sig.parameters.items():
|
||||
parameter = param[1]
|
||||
@ -138,7 +138,7 @@ class ServiceProvider(ServiceProviderABC):
|
||||
sb = ScopeBuilder(ServiceProvider(descriptors, self._configuration, self._database_context))
|
||||
return sb.build()
|
||||
|
||||
def get_service(self, service_type: T, *args, **kwargs) -> Optional[T]:
|
||||
def get_service(self, service_type: T, *args, **kwargs) -> Optional[R]:
|
||||
result = self._find_service(service_type)
|
||||
|
||||
if result is None:
|
||||
@ -157,7 +157,7 @@ class ServiceProvider(ServiceProviderABC):
|
||||
|
||||
return implementation
|
||||
|
||||
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[T]]:
|
||||
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[R]]:
|
||||
implementations = []
|
||||
|
||||
if typing.get_origin(service_type) == list:
|
||||
|
@ -4,7 +4,7 @@ from inspect import Signature, signature
|
||||
from typing import Optional, Type
|
||||
|
||||
from cpl_core.dependency_injection.scope_abc import ScopeABC
|
||||
from cpl_core.type import T
|
||||
from cpl_core.type import T, R
|
||||
|
||||
|
||||
class ServiceProviderABC(ABC):
|
||||
@ -21,7 +21,7 @@ class ServiceProviderABC(ABC):
|
||||
cls._provider = provider
|
||||
|
||||
@abstractmethod
|
||||
def build_by_signature(self, sig: Signature) -> list[T]:
|
||||
def build_by_signature(self, sig: Signature) -> list[R]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
@ -61,7 +61,7 @@ class ServiceProviderABC(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_service(self, instance_type: T, *args, **kwargs) -> Optional[T]:
|
||||
def get_service(self, instance_type: T, *args, **kwargs) -> Optional[R]:
|
||||
r"""Returns instance of given type
|
||||
|
||||
Parameter
|
||||
@ -76,7 +76,7 @@ class ServiceProviderABC(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[T]]:
|
||||
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[R]]:
|
||||
r"""Returns instance of given type
|
||||
|
||||
Parameter
|
||||
|
@ -1,3 +1,4 @@
|
||||
from typing import TypeVar
|
||||
|
||||
T = TypeVar("T")
|
||||
R = TypeVar("R")
|
||||
|
Loading…
Reference in New Issue
Block a user