Monday, 11 March 2024

Service Locator: Quick and Dirty Prototype

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

Parse Wikipedia dump

""" This module processes Wikipedia dump files by extracting individual articles and parsing them into a structured format, ...