""" Part of the bsie module. A copy of the license is provided with the project. Author: Matthias Baumgartner, 2022 """ # standard imports import typing # bsie imports from bsie.utils import bsfs, errors, safe_load, unpack_qualified_name # inner-module imports from . import base # exports __all__: typing.Sequence[str] = ( 'ReaderBuilder', ) ## code ## class ReaderBuilder(): """Build `bsie.base.Reader` instances. Readers are defined via their qualified class name (e.g., bsie.reader.path.Path) and optional keyword arguments that are passed to the constructor via the *kwargs* argument (name as key, kwargs as value). The ReaderBuilder keeps a cache of previously built reader instances, as they are anyway built with identical keyword arguments. """ # keyword arguments _kwargs: typing.Dict[str, typing.Dict[str, typing.Any]] # cached readers _cache: typing.Dict[str, base.Reader] def __init__( self, kwargs: typing.Optional[typing.Dict[str, typing.Dict[str, typing.Any]]] = None): if kwargs is None: kwargs = {} self._kwargs = kwargs self._cache = {} def build(self, name: str) -> base.Reader: """Return an instance for the qualified class name.""" # return cached instance if name in self._cache: return self._cache[name] # check name and get module/class components module_name, class_name = unpack_qualified_name(name) # import reader class cls = safe_load(module_name, class_name) # get kwargs kwargs = self._kwargs.get(name, {}) if not isinstance(kwargs, dict): raise TypeError(f'expected a kwargs dict, found {bsfs.typename(kwargs)}') try: # build, cache, and return instance obj = cls(**kwargs) # cache instance self._cache[name] = obj # return instance return obj except Exception as err: raise errors.BuilderError(f'failed to build reader {name} due to {bsfs.typename(err)}: {err}') from err ## EOF ##