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

Merge branch 'master' into 2023.4

Browse Source 
# Conflicts:
#	src/cpl_cli/.cpl/__init__.py
#	src/cpl_cli/__init__.py
#	src/cpl_cli/_templates/__init__.py
#	src/cpl_cli/_templates/build/__init__.py
#	src/cpl_cli/_templates/publish/__init__.py
#	src/cpl_cli/abc/__init__.py
#	src/cpl_cli/command/__init__.py
#	src/cpl_cli/configuration/__init__.py
#	src/cpl_cli/cpl-cli.json
#	src/cpl_cli/helper/__init__.py
#	src/cpl_cli/live_server/__init__.py
#	src/cpl_cli/migrations/__init__.py
#	src/cpl_cli/migrations/base/__init__.py
#	src/cpl_cli/migrations/service/__init__.py
#	src/cpl_cli/publish/__init__.py
#	src/cpl_cli/source_creator/__init__.py
#	src/cpl_cli/validators/__init__.py
#	src/cpl_core/__init__.py
#	src/cpl_core/application/__init__.py
#	src/cpl_core/configuration/__init__.py
#	src/cpl_core/console/__init__.py
#	src/cpl_core/cpl-core.json
#	src/cpl_core/database/__init__.py
#	src/cpl_core/database/connection/__init__.py
#	src/cpl_core/database/context/__init__.py
#	src/cpl_core/dependency_injection/__init__.py
#	src/cpl_core/dependency_injection/service_provider.py
#	src/cpl_core/dependency_injection/service_provider_abc.py
#	src/cpl_core/environment/__init__.py
#	src/cpl_core/logging/__init__.py
#	src/cpl_core/mailing/__init__.py
#	src/cpl_core/pipes/__init__.py
#	src/cpl_core/time/__init__.py
#	src/cpl_core/utils/__init__.py
#	src/cpl_discord/.cpl/__init__.py
#	src/cpl_discord/.cpl/schematic_discord_event.py
#	src/cpl_discord/__init__.py
#	src/cpl_discord/application/__init__.py
#	src/cpl_discord/command/__init__.py
#	src/cpl_discord/configuration/__init__.py
#	src/cpl_discord/container/__init__.py
#	src/cpl_discord/cpl-discord.json
#	src/cpl_discord/events/__init__.py
#	src/cpl_discord/helper/__init__.py
#	src/cpl_discord/service/__init__.py
#	src/cpl_discord/service/discord_bot_service.py
#	src/cpl_discord/service/discord_bot_service_abc.py
#	src/cpl_discord/service/discord_service.py
#	src/cpl_query/__init__.py
#	src/cpl_query/base/__init__.py
#	src/cpl_query/base/queryable_abc.py
#	src/cpl_query/cpl-query.json
#	src/cpl_query/enumerable/__init__.py
#	src/cpl_query/extension/__init__.py
#	src/cpl_query/iterable/__init__.py
This commit is contained in:
Sven Heidemann
2023-04-04 13:29:45 +02:00
parent 87d023ef7f 43710717c9
commit 01bc746092
14 changed files with 414 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,7 +18,8 @@ from cpl_core.type import T
class ServiceProvider(ServiceProviderABC):
r"""Provider for the services
Parameter:
Parameter
---------
service_descriptors: list[:class:`cpl_core.dependency_injection.service_descriptor.ServiceDescriptor`]
Descriptor of the service
config: :class:`cpl_core.configuration.configuration_abc.ConfigurationABC`
@@ -27,12 +28,7 @@ class ServiceProvider(ServiceProviderABC):
Database representation
"""
def __init__(
self,
service_descriptors: list[ServiceDescriptor],
config: ConfigurationABC,
db_context: Optional[DatabaseContextABC],
):
def __init__(self, service_descriptors: list[ServiceDescriptor], config: ConfigurationABC, db_context: Optional[DatabaseContextABC]):
ServiceProviderABC.__init__(self)
self._service_descriptors: list[ServiceDescriptor] = service_descriptors
@@ -49,9 +45,7 @@ class ServiceProvider(ServiceProviderABC):
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.service_type == parameter.annotation or issubclass(descriptor.service_type, parameter.annotation):
if descriptor.implementation is not None:
return descriptor.implementation
@@ -63,7 +57,7 @@ class ServiceProvider(ServiceProviderABC):
# raise Exception(f'Service {parameter.annotation} not found')
def _get_services(self, t: type) -> list[Optional[object]]:
def _get_services(self, t: type, *args, **kwargs) -> list[Optional[object]]:
implementations = []
for descriptor in self._service_descriptors:
if descriptor.service_type == t or issubclass(descriptor.service_type, t):
@@ -71,7 +65,7 @@ class ServiceProvider(ServiceProviderABC):
implementations.append(descriptor.implementation)
continue
implementation = self.build_service(descriptor.service_type)
implementation = self.build_service(descriptor.service_type, *args, **kwargs)
if descriptor.lifetime == ServiceLifetimeEnum.singleton:
descriptor.implementation = implementation
@@ -83,7 +77,8 @@ class ServiceProvider(ServiceProviderABC):
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != "self" and parameter.annotation != Parameter.empty:
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if typing.get_origin(parameter.annotation) == list:
params.append(self._get_services(typing.get_args(parameter.annotation)[0]))
@@ -107,7 +102,7 @@ class ServiceProvider(ServiceProviderABC):
return params
def build_service(self, service_type: type) -> object:
def build_service(self, service_type: type, *args, **kwargs) -> 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,18 +115,16 @@ class ServiceProvider(ServiceProviderABC):
sig = signature(service_type.__init__)
params = self.build_by_signature(sig)
return service_type(*params)
return service_type(*params, *args, **kwargs)
def set_scope(self, scope: ScopeABC):
self._scope = scope
def create_scope(self) -> ScopeABC:
sb = ScopeBuilder(
ServiceProvider(copy.deepcopy(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: T) -> Optional[T]:
def get_service(self, service_type: T, *args, **kwargs) -> Optional[T]:
result = self._find_service(service_type)
if result is None:
@@ -140,21 +133,17 @@ class ServiceProvider(ServiceProviderABC):
if result.implementation is not None:
return result.implementation
implementation = self.build_service(service_type)
if (
result.lifetime == ServiceLifetimeEnum.singleton
or result.lifetime == ServiceLifetimeEnum.scoped
and self._scope is not None
):
implementation = self.build_service(service_type, *args, **kwargs)
if result.lifetime == ServiceLifetimeEnum.singleton or result.lifetime == ServiceLifetimeEnum.scoped and self._scope is not None:
result.implementation = implementation
return implementation
def get_services(self, service_type: T) -> list[Optional[T]]:
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[T]]:
implementations = []
if typing.get_origin(service_type) != list:
raise Exception(f"Invalid type {service_type}! Expected list of type")
raise Exception(f'Invalid type {service_type}! Expected list of type')
implementations.extend(self._get_services(typing.get_args(service_type)[0]))

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

@@ -10,14 +10,14 @@ from cpl_core.type import T
class ServiceProviderABC(ABC):
r"""ABC for the class :class:`cpl_core.dependency_injection.service_provider.ServiceProvider`"""
_provider: Optional["ServiceProviderABC"] = None
_provider: Optional['ServiceProviderABC'] = None
@abstractmethod
def __init__(self):
pass
@classmethod
def set_global_provider(cls, provider: "ServiceProviderABC"):
def set_global_provider(cls, provider: 'ServiceProviderABC'):
cls._provider = provider
@abstractmethod
@@ -25,14 +25,16 @@ class ServiceProviderABC(ABC):
pass
@abstractmethod
def build_service(self, service_type: type) -> object:
def build_service(self, service_type: type, *args, **kwargs) -> object:
r"""Creates instance of given type
Parameter:
Parameter
---------
instance_type: :class:`type`
The type of the searched instance
Returns:
Returns
-------
Object of the given type
"""
pass
@@ -41,7 +43,8 @@ class ServiceProviderABC(ABC):
def set_scope(self, scope: ScopeABC):
r"""Sets the scope of service provider
Parameter:
Parameter
---------
Object of type :class:`cpl_core.dependency_injection.scope_abc.ScopeABC`
Service scope
"""
@@ -51,33 +54,38 @@ class ServiceProviderABC(ABC):
def create_scope(self) -> ScopeABC:
r"""Creates a service scope
Returns:
Returns
-------
Object of type :class:`cpl_core.dependency_injection.scope_abc.ScopeABC`
"""
pass
@abstractmethod
def get_service(self, instance_type: Type[T]) -> Optional[T]:
def get_service(self, instance_type: T, *args, **kwargs) -> Optional[T]:
r"""Returns instance of given type
Parameter:
Parameter
---------
instance_type: :class:`cpl_core.type.T`
The type of the searched instance
Returns:
Returns
-------
Object of type Optional[:class:`cpl_core.type.T`]
"""
pass
@abstractmethod
def get_services(self, service_type: T) -> list[Optional[T]]:
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[T]]:
r"""Returns instance of given type
Parameter:
Parameter
---------
instance_type: :class:`cpl_core.type.T`
The type of the searched instance
Returns:
Returns
-------
Object of type list[Optional[:class:`cpl_core.type.T`]
"""
pass
@@ -86,10 +94,12 @@ class ServiceProviderABC(ABC):
def inject(cls, f=None):
r"""Decorator to allow injection into static and class methods
Parameter:
Parameter
---------
f: Callable
Returns:
Returns
-------
function
"""
if f is None:
@@ -98,9 +108,9 @@ class ServiceProviderABC(ABC):
@functools.wraps(f)
def inner(*args, **kwargs):
if cls._provider is None:
raise Exception(f"{cls.__name__} not build!")
raise Exception(f'{cls.__name__} not build!')
injection = cls._provider.build_by_signature(signature(f))
return f(*injection, *args, **kwargs)
injection = [x for x in cls._provider.build_by_signature(signature(f)) if x is not None]
return f(*args, *injection, **kwargs)
return inner