from __future__ import annotations
from typing import TypeVar, Type
import inspect
import typing
T = TypeVar('T')
class Services:
def __init__(self):
self._singletons = {}
self._factories = {}
def singleton(self, type, service):
self._singletons[type] = service
type.instance = lambda: self.get(type)
def factory(self, type, factory):
self._factories[type] = factory
type.instance = lambda: self.get(type)
def get(self, type: Type[T]) -> T:
if type in self._singletons:
return self._singletons[type]
if type in self._factories:
return self._factories[type](self)
raise Exception(f'No service found for {type}')
def __call__(self, type: Type[T]) -> T:
return self.get(type)
def __getitem__(self, type: Type[T]) -> T:
return self.get(type)
def __setitem__(self, type: Type[T], service: T):
# if is lambda make it a factory or else a singleton
if inspect.isfunction(service):
self.factory(type, service)
else:
self.singleton(type, service)
def contains(self, type: Type[T]) -> bool:
return type in self._singletons or type in self._factories
def new(self, type: Type[T], **kwargs) -> T:
if self.contains(type):
return self.get(type)
return self.call(type, **kwargs)
def call(self, callable, **kwargs):
signature = inspect.signature(callable)
params = dict(signature.parameters)
if 'self' in params:
del params['self']
fn_kwargs = {}
type_hints = typing.get_type_hints(callable)
for param_name in params:
if param_name in kwargs:
fn_kwargs[param_name] = kwargs[param_name]
else:
required_type = type_hints[param_name]
instance = self.get(required_type)
fn_kwargs[param_name] = instance
return callable(**fn_kwargs)
services = Services()
@click.command()
def cli():
services[EvaluationRepository] = SqliteEvaluationRepository(session)
services[BenchmarkRepository] = SqliteBenchmarkRepository(session)
services[RetrieverRepository] = SqliteRetrieverRepository(session)
def __init__(self, projects_path: Path):
self.retriever_repository = services[RetrieverRepository]
self.projects_path = projects_path
self.evaluation_repository = services[EvaluationRepository]
self.benchmark_repository = services[BenchmarkRepository]
Conclusion: Did not use it as it removes seams, which makes refactoring in the future more complicated (Working effectively with legacy code), which decreases changability - my metric for good code.
No comments:
Post a Comment