Skip to content

plugins

Central place for convenient access to all registered plugin groups.

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

from metador_core.plugins import schemas

PGPluginGroup

Bases: ObjectProxy

PluginGroup plugin group.

This wrapper returns instances of other loaded plugin groups.

In the esoteric case that you need to access the actual plugingroup class that gives out classes instead of instances (like all other plugingroups), request the "plugingroup" plugingroup. But usually you will not want this.

Source code in src/metador_core/plugins.py
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
class PGPluginGroup(wrapt.ObjectProxy):
    """PluginGroup plugin group.

    This wrapper returns instances of other loaded plugin groups.

    In the esoteric case that you need to access the actual plugingroup class
    that gives out *classes* instead of instances (like all other plugingroups),
    request the "plugingroup" plugingroup. But usually you will not want this.
    """

    _self_groups: Dict[AnyPluginRef, PluginGroup]

    def __reset__(self):
        self._self_groups.clear()
        self.__init__()

    def __init__(self):
        # initialize the meta-plugingroup
        from .plugin.interface import _plugin_groups, create_pg

        create_pg(PluginGroup)
        pgpg_ref = AnyPluginRef(
            group=PG_GROUP_NAME,
            name=PluginGroup.Plugin.name,
            version=PluginGroup.Plugin.version,
        )

        # wire it up with this wrapper
        self._self_groups = _plugin_groups
        self.__wrapped__ = _plugin_groups[pgpg_ref]

    # ----
    def get(self, key, version=None):
        """Get a registered plugin group by name."""
        key_, vers = plugin_args(key, version)
        if key_ == self.name and (vers is None or vers == self.Plugin.version):
            return self
        try:
            if grp_cls := self.__wrapped__._get_unsafe(key_, vers):
                # now if the PG was not existing, it is + is stored in _self_groups
                return cast(S, self._self_groups.get(grp_cls.Plugin.ref()))
        except KeyError:
            return None

    def __getitem__(self, key) -> PluginGroup:
        # call wrapped '__getitem__' with this object to use its 'get'
        return PluginGroup.__getitem__(self, key)  # type: ignore

    def values(self):
        # same idea, this uses '__getitem__'
        return PluginGroup.values(self)

    def items(self):
        # same idea, this uses '__getitem__'
        return PluginGroup.items(self)

    def is_plugin(self, obj):
        return obj in self.values()

__init__

__init__()
Source code in src/metador_core/plugins.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def __init__(self):
    # initialize the meta-plugingroup
    from .plugin.interface import _plugin_groups, create_pg

    create_pg(PluginGroup)
    pgpg_ref = AnyPluginRef(
        group=PG_GROUP_NAME,
        name=PluginGroup.Plugin.name,
        version=PluginGroup.Plugin.version,
    )

    # wire it up with this wrapper
    self._self_groups = _plugin_groups
    self.__wrapped__ = _plugin_groups[pgpg_ref]

get

get(key, version=None)

Get a registered plugin group by name.

Source code in src/metador_core/plugins.py
48
49
50
51
52
53
54
55
56
57
58
def get(self, key, version=None):
    """Get a registered plugin group by name."""
    key_, vers = plugin_args(key, version)
    if key_ == self.name and (vers is None or vers == self.Plugin.version):
        return self
    try:
        if grp_cls := self.__wrapped__._get_unsafe(key_, vers):
            # now if the PG was not existing, it is + is stored in _self_groups
            return cast(S, self._self_groups.get(grp_cls.Plugin.ref()))
    except KeyError:
        return None