Merge branch 'develop'HEADv0.23.03releasemain

1 files changed, 73 insertions, 0 deletions

diff --git a/bsie/reader/builder.py b/bsie/reader/builder.py
new file mode 100644
index 0000000..d32700b
--- /dev/null
+++ b/bsie/reader/builder.py
@@ -0,0 +1,73 @@
+
+# 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 ##