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