Improved application structure
All checks were successful
Build on push / prepare (push) Successful in 11s
Build on push / core (push) Successful in 22s
Build on push / query (push) Successful in 22s
Build on push / dependency (push) Successful in 20s
Build on push / database (push) Successful in 20s
Build on push / translation (push) Successful in 20s
Build on push / application (push) Successful in 22s
Build on push / mail (push) Successful in 23s
Build on push / auth (push) Successful in 16s

This commit is contained in:
2025-09-17 19:23:14 +02:00
parent 504dc5e188
commit 8aaba22940
33 changed files with 200 additions and 381 deletions

View File

@@ -1,8 +1,7 @@
from typing import Union, Type, Callable, Optional
from typing import Union, Type, Callable
from cpl.core.log.logger import Logger
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.pipes.pipe_abc import PipeABC
from cpl.core.typing import T, Service
from cpl.dependency.service_descriptor import ServiceDescriptor
from cpl.dependency.service_lifetime_enum import ServiceLifetimeEnum
@@ -46,29 +45,6 @@ class ServiceCollection:
return self
def add_module(self, module: str | object):
if not isinstance(module, str):
module = module.__name__
if module not in self._modules:
raise ValueError(f"Module {module} not found")
self._modules[module](self)
# def add_mysql(self, db_context_type: Type[DatabaseContextABC], db_settings: DatabaseSettings):
# self.add_singleton(DatabaseContextABC, db_context_type)
# self._database_context = self.build_service_provider().get_service(DatabaseContextABC)
# self._database_context.connect(db_settings)
def add_logging(self):
self.add_transient(LoggerABC, Logger)
return self
def add_pipes(self):
for pipe in PipeABC.__subclasses__():
self.add_transient(PipeABC, pipe)
return self
def add_singleton(self, service_type: T, service: Service = None):
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.singleton, service)
return self
@@ -81,7 +57,20 @@ class ServiceCollection:
self._add_descriptor_by_lifetime(service_type, ServiceLifetimeEnum.transient, service)
return self
def build_service_provider(self) -> ServiceProviderABC:
def build(self) -> ServiceProviderABC:
sp = ServiceProvider(self._service_descriptors)
ServiceProviderABC.set_global_provider(sp)
return sp
def add_module(self, module: str | object):
if not isinstance(module, str):
module = module.__name__
if module not in self._modules:
raise ValueError(f"Module {module} not found")
self._modules[module](self)
def add_logging(self):
self.add_transient(LoggerABC, Logger)
return self

View File

@@ -136,7 +136,7 @@ class ServiceProvider(ServiceProviderABC):
else:
descriptors.append(copy.deepcopy(descriptor))
sb = ScopeBuilder(ServiceProvider(descriptors, self._database_context))
sb = ScopeBuilder(ServiceProvider(descriptors))
return sb.build()
def get_service(self, service_type: T, *args, **kwargs) -> Optional[R]:

View File

@@ -1,7 +1,7 @@
import functools
from abc import abstractmethod, ABC
from inspect import Signature, signature
from typing import Optional
from typing import Optional, Type
from cpl.core.typing import T, R
from cpl.dependency.scope_abc import ScopeABC
@@ -74,7 +74,7 @@ class ServiceProviderABC(ABC):
"""
@abstractmethod
def get_service(self, instance_type: T, *args, **kwargs) -> Optional[R]:
def get_service(self, instance_type: Type[T], *args, **kwargs) -> Optional[T]:
r"""Returns instance of given type
Parameter
@@ -88,7 +88,7 @@ class ServiceProviderABC(ABC):
"""
@abstractmethod
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[R]]:
def get_services(self, service_type: Type[T], *args, **kwargs) -> list[Optional[T]]:
r"""Returns instance of given type
Parameter