2021.4.2 #12

Merged
edraft merged 22 commits from 2021.4.2 into 2021.4 2021-03-29 15:48:59 +02:00
12 changed files with 375 additions and 173 deletions
Showing only changes of commit c650e87443 - Show all commits

View File

@ -6,6 +6,7 @@ from cpl.application.application_runtime import ApplicationRuntime
from cpl.application.startup_abc import StartupABC
edraft marked this conversation as resolved Outdated

Direci import change to from cpl.configuration.configuration import Configuration

Direci import change to from cpl.configuration.configuration import Configuration
from cpl.configuration import Configuration
from cpl.dependency_injection import ServiceProvider
from cpl.dependency_injection.service_collection import ServiceCollection
class ApplicationBuilder(ApplicationBuilderABC):
@ -20,7 +21,7 @@ class ApplicationBuilder(ApplicationBuilderABC):
self._configuration = Configuration()
self._runtime = ApplicationRuntime()
self._services = ServiceProvider(self._configuration, self._runtime)
self._services = ServiceCollection(self._configuration, self._runtime)
def use_startup(self, startup: Type[StartupABC]):
"""
@ -39,4 +40,4 @@ class ApplicationBuilder(ApplicationBuilderABC):
self._startup.configure_configuration()
self._startup.configure_services()
return self._app(self._configuration, self._runtime, self._services)
return self._app(self._configuration, self._runtime, self._services.build_service_provider())

View File

@ -21,7 +21,7 @@ from collections import namedtuple
# imports:
from .service_abc import ServiceABC
from .service_provider import ServiceProvider
from .service_provider_old import ServiceProvider
from .service_provider_abc import ServiceProviderABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')

View File

@ -0,0 +1,67 @@
from typing import Union, Type, Callable, Optional
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.database.context import DatabaseContextABC
from cpl.dependency_injection.service_factory import ServiceFactory
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
from cpl.dependency_injection.service_provider import ServiceProvider
class ServiceCollection(ServiceCollectionABC):
def __init__(self, config: ConfigurationABC, runtime: ApplicationRuntimeABC):
ServiceCollectionABC.__init__(self)
self._configuration: ConfigurationABC = config
self._runtime: ApplicationRuntimeABC = runtime
self._database_context: Optional[DatabaseContextABC] = None
self._service_descriptors: list[ServiceDescriptor] = []
def _add_descriptor(self, service: Union[type, object], lifetime: ServiceLifetimeEnum):
found = False
for descriptor in self._service_descriptors:
if not isinstance(service, type):
service = type(service)
if descriptor.service_type == service:
found = True
if found:
service_type = service
if not isinstance(service, type):
service_type = type(service)
raise Exception(f'Service of type {service_type} already exists')
self._service_descriptors.append(ServiceDescriptor(service, lifetime))
def add_db_context(self, db_context: Type[DatabaseContextABC]):
pass
def add_singleton(self, service_type: Union[type, object], service: Union[type, object] = None):
if service is not None:
if isinstance(service, type):
service = service()
self._add_descriptor(service, ServiceLifetimeEnum.singleton)
else:
if isinstance(service_type, type):
service_type = service_type()
self._add_descriptor(service_type, ServiceLifetimeEnum.singleton)
def add_scoped(self, service_type: Type, service: Callable = None):
pass
def add_transient(self, service_type: Union[type], service: Union[type] = None):
if service is not None:
self._add_descriptor(service, ServiceLifetimeEnum.transient)
else:
self._add_descriptor(service_type, ServiceLifetimeEnum.transient)
def build_service_provider(self) -> ServiceProviderABC:
return ServiceProvider(ServiceFactory(self._service_descriptors, self._configuration, self._runtime))

View File

@ -0,0 +1,62 @@
from abc import abstractmethod, ABC
from collections import Callable
from typing import Type
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
class ServiceCollectionABC(ABC):
@abstractmethod
def __init__(self):
"""
ABC for service providing
"""
pass
@abstractmethod
def add_db_context(self, db_context: Type[DatabaseContextABC]):
"""
Adds database context
:param db_context:
:return:
"""
pass
@abstractmethod
def add_transient(self, service_type: Type, service: Callable = None):
"""
Adds a service with transient lifetime
:param service_type:
:param service:
:return:
"""
pass
@abstractmethod
def add_scoped(self, service_type: Type, service: Callable = None):
"""
Adds a service with scoped lifetime
:param service_type:
:param service:
:return:
"""
pass
@abstractmethod
def add_singleton(self, service_type: Type, service: Callable = None):
"""
Adds a service with singleton lifetime
:param service_type:
:param service:
:return:
"""
pass
@abstractmethod
def build_service_provider(self) -> ServiceProviderABC:
"""
Creates instance of the service provider
"""
pass

View File

@ -0,0 +1,33 @@
from typing import Union, Optional
from cpl.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
class ServiceDescriptor:
def __init__(self, implementation: Union[type, Optional[object]], lifetime: ServiceLifetimeEnum):
self._service_type = implementation
self._implementation = implementation
self._lifetime = lifetime
if not isinstance(implementation, type):
self._service_type = type(implementation)
else:
self._implementation = None
@property
def service_type(self) -> type:
return self._service_type
@property
def implementation(self) -> Union[type, Optional[object]]:
return self._implementation
@implementation.setter
def implementation(self, implementation: Union[type, Optional[object]]):
self._implementation = implementation
@property
def lifetime(self) -> ServiceLifetimeEnum:
return self._lifetime

View File

@ -0,0 +1,27 @@
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.dependency_injection.service_factory_abc import ServiceFactoryABC
class ServiceFactory(ServiceFactoryABC):
def __init__(self, service_descriptors: list[ServiceDescriptor], config: ConfigurationABC,
runtime: ApplicationRuntimeABC):
ServiceFactoryABC.__init__(self)
self._service_descriptors: list[ServiceDescriptor] = service_descriptors
self._configuration: ConfigurationABC = config
self._runtime: ApplicationRuntimeABC = runtime
@property
def service_descriptors(self) -> list[ServiceDescriptor]:
return self._service_descriptors
@property
def configuration(self) -> ConfigurationABC:
return self._configuration
@property
def runtime(self) -> ApplicationRuntimeABC:
return self._runtime

View File

@ -0,0 +1,23 @@
from abc import ABC, abstractmethod
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
class ServiceFactoryABC(ABC):
@abstractmethod
def __init__(self): pass
@property
@abstractmethod
def service_descriptors(self) -> list[ServiceDescriptor]: pass
@property
@abstractmethod
def configuration(self) -> ConfigurationABC: pass
@property
@abstractmethod
def runtime(self) -> ApplicationRuntimeABC: pass

View File

@ -0,0 +1,8 @@
from enum import Enum
class ServiceLifetimeEnum(Enum):
singleton = 0
scoped = 1 # not supported yet
transient = 2

View File

@ -1,122 +1,37 @@
from collections import Callable
from inspect import signature, Parameter
from typing import Type, Optional, Union
from typing import Optional
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_abc import ServiceABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.environment.environment_abc import ApplicationEnvironmentABC
from cpl.dependency_injection import ServiceProviderABC
from cpl.dependency_injection.service_descriptor import ServiceDescriptor
from cpl.dependency_injection.service_factory_abc import ServiceFactoryABC
from cpl.dependency_injection.service_lifetime_enum import ServiceLifetimeEnum
class ServiceProvider(ServiceProviderABC):
def __init__(self, config: ConfigurationABC, runtime: ApplicationRuntimeABC):
"""
Service for service providing
:param runtime:
"""
def __init__(self, service_factory: ServiceFactoryABC):
ServiceProviderABC.__init__(self)
self._configuration: ConfigurationABC = config
self._runtime: ApplicationRuntimeABC = runtime
self._database_context: Optional[DatabaseContextABC] = None
self._transient_services: dict[Type[ServiceABC], Callable[ServiceABC]] = {}
self._scoped_services: dict[Type[ServiceABC], Callable[ServiceABC]] = {}
self._singleton_services: dict[Type[ServiceABC], Callable[ServiceABC], ServiceABC] = {}
self._service_factory = service_factory
def _create_instance(self, service: Union[Callable[ServiceABC], ServiceABC]) -> Callable[ServiceABC]:
"""
Creates an instance of given type
:param service:
:return:
"""
sig = signature(service.__init__)
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if issubclass(parameter.annotation, ApplicationRuntimeABC):
params.append(self._runtime)
def _find_service(self, service_type: type) -> [ServiceDescriptor]:
for descriptor in self._service_factory.service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
return descriptor
elif issubclass(parameter.annotation, ApplicationEnvironmentABC):
params.append(self._configuration.environment)
return None
elif issubclass(parameter.annotation, DatabaseContextABC):
params.append(self._database_context)
def get_service(self, service_type: type) -> Optional[Callable[object]]:
result = self._find_service(service_type)
elif issubclass(parameter.annotation, ConfigurationModelABC):
params.append(self._configuration.get_configuration(parameter.annotation))
if result is None:
return None
elif issubclass(parameter.annotation, ConfigurationABC):
params.append(self._configuration)
if result.implementation is not None:
return result.implementation
elif issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
implementation = result.service_type()
if result.lifetime == ServiceLifetimeEnum.singleton:
result.implementation = implementation
else:
params.append(self.get_service(parameter.annotation))
return service(*params)
def add_db_context(self, db_context: Type[DatabaseContextABC]):
self._database_context = self._create_instance(db_context)
def get_db_context(self) -> Callable[DatabaseContextABC]:
return self._database_context
def add_transient(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
if service is None:
self._transient_services[service_type] = service_type
else:
self._transient_services[service_type] = service
def add_scoped(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
if service is None:
self._scoped_services[service_type] = service_type
else:
self._scoped_services[service_type] = service
def add_singleton(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
for known_service in self._singleton_services:
if type(known_service) == service_type:
raise Exception(f'Service with type {service_type} already exists')
if service is None:
self._singleton_services[service_type] = self._create_instance(service_type)
else:
self._singleton_services[service_type] = self._create_instance(service)
def get_service(self, instance_type: Type) -> Callable[ServiceABC]:
if issubclass(instance_type, ServiceProviderABC):
return self
for service in self._transient_services:
if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
return self._create_instance(self._transient_services[service])
for service in self._scoped_services:
if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
return self._create_instance(self._scoped_services[service])
for service in self._singleton_services:
if service == instance_type and isinstance(self._singleton_services[service], instance_type):
return self._singleton_services[service]
def remove_service(self, instance_type: Type[ServiceABC]):
for service in self._transient_services:
if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
del self._transient_services[service]
return
for service in self._scoped_services:
if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
del self._scoped_services[service]
return
for service in self._singleton_services:
if service == instance_type and isinstance(self._singleton_services[service], instance_type):
del self._singleton_services[service]
return
return implementation

View File

@ -2,7 +2,6 @@ from abc import abstractmethod, ABC
from collections import Callable
from typing import Type
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_abc import ServiceABC
@ -15,53 +14,6 @@ class ServiceProviderABC(ABC):
"""
pass
@abstractmethod
def add_db_context(self, db_context: Type[DatabaseContextABC]):
"""
Adds database context
:param db_context:
:return:
"""
pass
@abstractmethod
def get_db_context(self) -> Callable[DatabaseContextABC]:
""""
Returns database context
:return Callable[DatabaseContextABC]:
"""
pass
@abstractmethod
def add_transient(self, service_type: Type, service: Callable = None):
"""
Adds a service with transient lifetime
:param service_type:
:param service:
:return:
"""
pass
@abstractmethod
def add_scoped(self, service_type: Type, service: Callable = None):
"""
Adds a service with scoped lifetime
:param service_type:
:param service:
:return:
"""
pass
@abstractmethod
def add_singleton(self, service_type: Type, service: Callable = None):
"""
Adds a service with singleton lifetime
:param service_type:
:param service:
:return:
"""
pass
@abstractmethod
def get_service(self, instance_type: Type) -> Callable[ServiceABC]:
"""
@ -70,12 +22,3 @@ class ServiceProviderABC(ABC):
:return:
"""
pass
@abstractmethod
def remove_service(self, instance_type: type):
"""
Removes service
:param instance_type:
:return:
"""
pass

View File

@ -0,0 +1,122 @@
from collections import Callable
from inspect import signature, Parameter
from typing import Type, Optional, Union
from cpl.application.application_runtime_abc import ApplicationRuntimeABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.database.context.database_context_abc import DatabaseContextABC
from cpl.dependency_injection.service_abc import ServiceABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.environment.environment_abc import ApplicationEnvironmentABC
class ServiceProvider(ServiceProviderABC):
def __init__(self, config: ConfigurationABC, runtime: ApplicationRuntimeABC):
"""
Service for service providing
:param runtime:
"""
ServiceProviderABC.__init__(self)
self._configuration: ConfigurationABC = config
self._runtime: ApplicationRuntimeABC = runtime
self._database_context: Optional[DatabaseContextABC] = None
self._transient_services: dict[Type[ServiceABC], Callable[ServiceABC]] = {}
self._scoped_services: dict[Type[ServiceABC], Callable[ServiceABC]] = {}
self._singleton_services: dict[Type[ServiceABC], Callable[ServiceABC], ServiceABC] = {}
def _create_instance(self, service: Union[Callable[ServiceABC], ServiceABC]) -> Callable[ServiceABC]:
"""
Creates an instance of given type
:param service:
:return:
"""
sig = signature(service.__init__)
params = []
for param in sig.parameters.items():
parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty:
if issubclass(parameter.annotation, ApplicationRuntimeABC):
params.append(self._runtime)
elif issubclass(parameter.annotation, ApplicationEnvironmentABC):
params.append(self._configuration.environment)
elif issubclass(parameter.annotation, DatabaseContextABC):
params.append(self._database_context)
elif issubclass(parameter.annotation, ConfigurationModelABC):
params.append(self._configuration.get_configuration(parameter.annotation))
elif issubclass(parameter.annotation, ConfigurationABC):
params.append(self._configuration)
elif issubclass(parameter.annotation, ServiceProviderABC):
params.append(self)
else:
params.append(self.get_service(parameter.annotation))
return service(*params)
def add_db_context(self, db_context: Type[DatabaseContextABC]):
self._database_context = self._create_instance(db_context)
def get_db_context(self) -> Callable[DatabaseContextABC]:
return self._database_context
def add_transient(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
if service is None:
self._transient_services[service_type] = service_type
else:
self._transient_services[service_type] = service
def add_scoped(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
if service is None:
self._scoped_services[service_type] = service_type
else:
self._scoped_services[service_type] = service
def add_singleton(self, service_type: Type[ServiceABC], service: Callable[ServiceABC] = None):
for known_service in self._singleton_services:
if type(known_service) == service_type:
raise Exception(f'Service with type {service_type} already exists')
if service is None:
self._singleton_services[service_type] = self._create_instance(service_type)
else:
self._singleton_services[service_type] = self._create_instance(service)
def get_service(self, instance_type: Type) -> Callable[ServiceABC]:
if issubclass(instance_type, ServiceProviderABC):
return self
for service in self._transient_services:
if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
return self._create_instance(self._transient_services[service])
for service in self._scoped_services:
if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
return self._create_instance(self._scoped_services[service])
for service in self._singleton_services:
if service == instance_type and isinstance(self._singleton_services[service], instance_type):
return self._singleton_services[service]
def remove_service(self, instance_type: Type[ServiceABC]):
for service in self._transient_services:
if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
del self._transient_services[service]
return
for service in self._scoped_services:
if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
del self._scoped_services[service]
return
for service in self._singleton_services:
if service == instance_type and isinstance(self._singleton_services[service], instance_type):
del self._singleton_services[service]
return

View File

@ -3,6 +3,7 @@ from cpl.application.startup_abc import StartupABC
from cpl.configuration.configuration_abc import ConfigurationABC
from cpl.database.context.database_context import DatabaseContext
from cpl.database.database_settings import DatabaseSettings
from cpl.dependency_injection.service_collection_abc import ServiceCollectionABC
from cpl.dependency_injection.service_provider_abc import ServiceProviderABC
from cpl.logging.logger_service import Logger
from cpl.logging.logger_abc import LoggerABC
@ -13,12 +14,13 @@ from cpl.utils.credential_manager import CredentialManager
class Startup(StartupABC):
def __init__(self, config: ConfigurationABC, runtime: ApplicationRuntimeABC, services: ServiceProviderABC):
def __init__(self, config: ConfigurationABC, runtime: ApplicationRuntimeABC, services: ServiceCollectionABC):
StartupABC.__init__(self)
self._configuration = config
self._application_runtime = runtime
self._services = services
print(self._services)
def configure_configuration(self) -> ConfigurationABC:
self._configuration.add_environment_variables('PYTHON_')
@ -32,12 +34,11 @@ class Startup(StartupABC):
def configure_services(self) -> ServiceProviderABC:
# Create and connect to database
db_settings: DatabaseSettings = self._configuration.get_configuration(DatabaseSettings)
self._services.add_db_context(DatabaseContext)
db: DatabaseContext = self._services.get_db_context()
db.connect(CredentialManager.build_string(db_settings.connection_string, db_settings.credentials))
# db_settings: DatabaseSettings = self._configuration.get_configuration(DatabaseSettings)
# self._services.add_db_context(DatabaseContext)
# db.connect(CredentialManager.build_string(db_settings.connection_string, db_settings.credentials))
self._services.add_singleton(LoggerABC, Logger)
self._services.add_singleton(EMailClientABC, EMailClient)
return self._services
return self._services.build_service_provider()