1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
"""
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 ##
|
