More efficient wrapped logger by getting service type not service
Some checks failed
Test before pr merge / test-lint (pull_request) Failing after 6s
Build on push / prepare (push) Successful in 10s
Build on push / core (push) Successful in 17s
Build on push / query (push) Successful in 24s
Build on push / dependency (push) Successful in 17s
Build on push / application (push) Successful in 15s
Build on push / translation (push) Successful in 15s
Build on push / database (push) Successful in 19s
Build on push / mail (push) Successful in 19s
Build on push / auth (push) Successful in 14s
Build on push / api (push) Successful in 14s

This commit is contained in:
2025-09-24 08:28:50 +02:00
parent e296c0992b
commit c71a3df62c
3 changed files with 52 additions and 7 deletions

View File

@@ -1,7 +1,8 @@
import inspect import inspect
from typing import Type
from cpl.core.log import LoggerABC, LogLevel from cpl.core.log import LoggerABC, LogLevel
from cpl.core.typing import Messages, Source from cpl.core.typing import Messages
from cpl.dependency.service_provider_abc import ServiceProviderABC from cpl.dependency.service_provider_abc import ServiceProviderABC
@@ -11,18 +12,20 @@ class WrappedLogger(LoggerABC):
LoggerABC.__init__(self) LoggerABC.__init__(self)
assert file_prefix is not None and file_prefix != "", "file_prefix must be a non-empty string" assert file_prefix is not None and file_prefix != "", "file_prefix must be a non-empty string"
t_logger = ServiceProviderABC.get_global_service(LoggerABC)
self._t_logger = type(t_logger) if t_logger is not None else None
self._source = None self._source = None
self._file_prefix = file_prefix self._file_prefix = file_prefix
self._set_logger() self._set_logger()
def _set_logger(self): @ServiceProviderABC.inject
if self._t_logger is None: def _set_logger(self, services: ServiceProviderABC):
from cpl.core.log import Logger
t_logger: Type[Logger] = services.get_service_type(LoggerABC)
if t_logger is None:
raise Exception("No LoggerABC service registered in ServiceProviderABC") raise Exception("No LoggerABC service registered in ServiceProviderABC")
self._logger = self._t_logger(self._source, self._file_prefix) self._logger = t_logger(self._source, self._file_prefix)
def set_level(self, level: LogLevel): def set_level(self, level: LogLevel):
self._logger.set_level(level) self._logger.set_level(level)

View File

@@ -1,7 +1,7 @@
import copy import copy
import typing import typing
from inspect import signature, Parameter, Signature from inspect import signature, Parameter, Signature
from typing import Optional from typing import Optional, Type
from cpl.core.configuration import Configuration from cpl.core.configuration import Configuration
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
@@ -158,6 +158,13 @@ class ServiceProvider(ServiceProviderABC):
return implementation return implementation
def get_service_type(self, service_type: Type[T]) -> Optional[Type[T]]:
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
return descriptor.service_type
return None
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[R]]: def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[R]]:
implementations = [] implementations = []
@@ -167,3 +174,10 @@ class ServiceProvider(ServiceProviderABC):
implementations.extend(self._get_services(service_type)) implementations.extend(self._get_services(service_type))
return implementations return implementations
def get_service_types(self, service_type: Type[T]) -> list[Type[T]]:
types = []
for descriptor in self._service_descriptors:
if descriptor.service_type == service_type or issubclass(descriptor.service_type, service_type):
types.append(descriptor.service_type)
return types

View File

@@ -85,6 +85,20 @@ class ServiceProviderABC(ABC):
Object of type Optional[:class:`cpl.core.type.T`] Object of type Optional[:class:`cpl.core.type.T`]
""" """
@abstractmethod
def get_service_type(self,instance_type: Type[T]) -> Optional[Type[T]]:
r"""Returns the registered service type for loggers
Parameter
---------
instance_type: :class:`cpl.core.type.T`
The type of the searched instance
Returns
-------
Object of type Optional[:class:`type`]
"""
@abstractmethod @abstractmethod
def get_services(self, service_type: Type[T], *args, **kwargs) -> list[Optional[T]]: def get_services(self, service_type: Type[T], *args, **kwargs) -> list[Optional[T]]:
r"""Returns instance of given type r"""Returns instance of given type
@@ -99,6 +113,20 @@ class ServiceProviderABC(ABC):
Object of type list[Optional[:class:`cpl.core.type.T`] Object of type list[Optional[:class:`cpl.core.type.T`]
""" """
@abstractmethod
def get_service_types(self, service_type: Type[T]) -> list[Type[T]]:
r"""Returns all registered service types
Parameter
---------
service_type: :class:`cpl.core.type.T`
The type of the searched instance
Returns
-------
Object of type list[:class:`type`]
"""
@classmethod @classmethod
def inject(cls, f=None): def inject(cls, f=None):
r"""Decorator to allow injection into static and class methods r"""Decorator to allow injection into static and class methods