Aggregate provider (aka use Selector as FactoryAggregate) #530
Description
I found a selector provider, but I think it doesn't satisfy my needs. I woud like to write a CLI script (i'm using click library) which reads configs of different types (json / yaml). I want users to specify a config type in an input and choose an appropriate reader with specified value. Also I want a click to validate an input, so it can show error if user specifies invalid / unknown type of the config. The last thing can be made by using a click.argument with type click.Choice.
Below is a code with Selector provider usage.
import abc import typing as t from contextlib import closing from pathlib import Path import click from dependency_injector.containers import DeclarativeContainer from dependency_injector.providers import Configuration, Selector, Singleton class Reader(metaclass=abc.ABCMeta): @abc.abstractmethod def read(self, source: t.TextIO) -> object: raise NotImplementedError class YamlReader(Reader): def read(self, source: t.TextIO) -> object: return "yaml config" class JsonReader(Reader): def read(self, source: t.TextIO) -> object: return "json config" class CLIContainer(DeclarativeContainer): # I have to declare a mapping with all available readers separately, to use it in `click.Choice` type. READERS = { "yaml": Singleton(YamlReader), "json": Singleton(JsonReader), } # I have to declare a configuration to use it with `Selector` config = Configuration() reader = Selector(config.reader_type, **READERS) @click.command("cli") @click.argument("type", type=click.Choice(sorted(CLIContainer.READERS))) # pass all available reader types. @click.argument("config", type=click.Path(exists=True, path_type=Path)) @click.pass_context def cli(context: click.Context, type: str, config: Path) -> None: container = context.obj = CLIContainer() # I have to specify a value in a config, instead of passing it to a `container.reader` provider. Also I have to # set a value in a config via a name of the config value is used in a selector provider. container.config.set("reader_type", type) reader = container.reader() click.echo(reader.read(context.with_resource(closing(config.open("r"))))) if __name__ == "__main__": cli()What I want is to write a less code and to use mypy checks / IDE suggestions to avoid mistakes. That's why a came up with an implementation of KeySelector provider in my providers.py module. See the code below.
import typing as t from dependency_injector.providers import Callable, Provider, deepcopy T = t.TypeVar("T") class KeySelector(Provider, t.Generic[T]): __slots__ = ( "__providers_by_key", "__inner", ) def __init__(self, providers_by_key: t.Mapping[str, Provider[T]], *args: t.Any, **kwargs: t.Any) -> None: self.__providers_by_key = providers_by_key self.__inner = Callable(self.__providers_by_key.get, *args, **kwargs) super().__init__() def __deepcopy__(self, memo: t.Dict[int, t.Any]) -> "KeySelector": copied = memo.get(id(self)) if copied is not None: return copied copied = self.__class__( deepcopy(self.__providers_by_key, memo), *deepcopy(self.__inner.args, memo), **deepcopy(self.__inner.kwargs, memo), ) self._copy_overridings(copied, memo) return copied @property def related(self) -> t.Iterable[Provider]: """Return related providers generator.""" yield self.__inner yield from super().related def _provide(self, args: t.Sequence[t.Any], kwargs: t.Mapping[str, t.Any]) -> T: key, *args = args return self.__inner(key)(*args, **kwargs) def keys(self) -> t.Collection[str]: return self.__providers_by_key.keys()Then I can use it in my CLI as I want.
import abc import typing as t from contextlib import closing from pathlib import Path import click from dependency_injector.containers import DeclarativeContainer from dependency_injector.providers import Singleton from .providers import KeySelector class Reader(metaclass=abc.ABCMeta): @abc.abstractmethod def read(self, source: t.TextIO) -> object: raise NotImplementedError class YamlReader(Reader): def read(self, source: t.TextIO) -> object: return "yaml config" class JsonReader(Reader): def read(self, source: t.TextIO) -> object: return "json config" class CLIContainer(DeclarativeContainer): reader = KeySelector({ "yaml": Singleton(YamlReader), "json": Singleton(JsonReader), }) @click.command("cli") @click.argument("type", type=click.Choice(sorted(CLIContainer.reader.keys()))) # pass all available reader types. @click.argument("config", type=click.Path(exists=True, path_type=Path)) @click.pass_context def cli(context: click.Context, type: str, config: Path) -> None: container = context.obj = CLIContainer() # Now I can get an instance by passing a key to my provider. It's more simple to me. reader = container.reader(type) click.echo(reader.read(context.with_resource(closing(config.open("r"))))) if __name__ == "__main__": cli()Maybe my suggestion can be enhanced and can be added to a library, I guess? 😃
P.S. For know I don't know how to implement override method for KeySelector provider. Maybe KeySelector may just provide an interface of Mapping type, so all providers can be overriden directly via a key selector["json"].override(...).