Coverage for src/metador_core/plugins.py: 100%

1"""Central place for convenient access to all registered plugin groups. 

2 

3For example, to access the `schema` plugingroup, you can use: 

4 

5`from metador_core.plugins import schemas` 

6""" 

7from typing import TYPE_CHECKING, Dict, List, TypeVar, cast 

8 

9import wrapt 

10 

11from .plugin.interface import PG_GROUP_NAME, AnyPluginRef, PluginGroup, plugin_args 

12 

13S = TypeVar("S", bound=PluginGroup) 

14 

15 

16class PGPluginGroup(wrapt.ObjectProxy): 

17 """PluginGroup plugin group. 

18 

19 This wrapper returns instances of other loaded plugin groups. 

20 

21 In the esoteric case that you need to access the actual plugingroup class 

22 that gives out *classes* instead of instances (like all other plugingroups), 

23 request the "plugingroup" plugingroup. But usually you will not want this. 

24 """ 

25 

26 _self_groups: Dict[AnyPluginRef, PluginGroup] 

27 

28 def __reset__(self): 

29 self._self_groups.clear() 

30 self.__init__() 

31 

32 def __init__(self): 

33 # initialize the meta-plugingroup 

34 from .plugin.interface import _plugin_groups, create_pg 

35 

36 create_pg(PluginGroup) 

37 pgpg_ref = AnyPluginRef( 

38 group=PG_GROUP_NAME, 

39 name=PluginGroup.Plugin.name, 

40 version=PluginGroup.Plugin.version, 

41 ) 

42 

43 # wire it up with this wrapper 

44 self._self_groups = _plugin_groups 

45 self.__wrapped__ = _plugin_groups[pgpg_ref] 

46 

47 # ---- 

48 def get(self, key, version=None): 

49 """Get a registered plugin group by name.""" 

50 key_, vers = plugin_args(key, version) 

51 if key_ == self.name and (vers is None or vers == self.Plugin.version): 

52 return self 

53 try: 

54 if grp_cls := self.__wrapped__._get_unsafe(key_, vers): 

55 # now if the PG was not existing, it is + is stored in _self_groups 

56 return cast(S, self._self_groups.get(grp_cls.Plugin.ref())) 

57 except KeyError: 

58 return None 

59 

60 def __getitem__(self, key) -> PluginGroup: 

61 # call wrapped '__getitem__' with this object to use its 'get' 

62 return PluginGroup.__getitem__(self, key) # type: ignore 

63 

64 def values(self): 

65 # same idea, this uses '__getitem__' 

66 return PluginGroup.values(self) 

67 

68 def items(self): 

69 # same idea, this uses '__getitem__' 

70 return PluginGroup.items(self) 

71 

72 def is_plugin(self, obj): 

73 return obj in self.values() 

74 

75 

76# access to available plugin groups: 

77 

78plugingroups: PGPluginGroup = PGPluginGroup() 

79 

80# help mypy (obviously only for groups in this package): 

81# NOTE: this would be better: https://github.com/python/mypy/issues/13643 

82if TYPE_CHECKING: # pragma: no cover 

83 from .harvester import PGHarvester 

84 from .packer import PGPacker 

85 from .schema.pg import PGSchema 

86 from .widget import PGWidget 

87 

88 schemas: PGSchema 

89 harvesters: PGHarvester 

90 widgets: PGWidget 

91 packers: PGPacker 

92 

93# ---- 

94# Now some magic to lift all other groups to module level, 

95# this allows to import like: from metador_core.plugins import schemas 

96 

97# define what to import with * 

98__all__ = list(sorted(map(lambda ref: f"{ref.name}s", plugingroups.keys()))) 

99 

100 

101def __dir__() -> List[str]: 

102 # show the existing plugin groups for tab completion in e.g. ipython 

103 return __all__ 

104 

105 

106def __getattr__(key: str): 

107 # get desired plugin group and add it as module attribute 

108 # (i.e. this is called at most once per group) 

109 if not isinstance(key, str) or key[-1] != "s": 

110 raise AttributeError(key) 

111 if group := plugingroups.get(key[:-1]): 

112 globals()["__annotations__"][key] = type(group) 

113 globals()[key] = group 

114 return group 

115 raise AttributeError(key)