Skip to content

Grapplers (Custom loading behavior)

grappler.grapplers

This module contains all of the grappler implementations that are provided with the package.

StaticGrappler(*objs: Tuple[Collection[str], Any], package: Optional[Package] = None) -> None

Bases: BasicGrappler[Dict[Plugin, Any]]

A grappler for loading "plugins" supplied by the host application.

This is provided as a useful tool to help modularize application code, so that application components can be loaded in the same way as plugins. To use this, supply an object as well as topics that each object implements to either __init__ or add_plugin, and the grappler will generate the appropriate Plugin tuples. The plugin.package is the same for every plugin yielded by an instance of this grappler. If a package argument is provided to the constructor, then this is used. Otherwise, a default internal package is used (StaticGrappler.internal_package).

Usage:

grappler = StaticGrappler(
    (["list", "of", "topics"], obj),
    ...
)
grappler.add_plugin(["topics", "list"], obj2)
Source code in grappler/grapplers/_static.py 
48
49
50
51
52
53
54
55
56
def __init__(
    self, *objs: Tuple[Collection[str], Any], package: Optional[Package] = None
) -> None:
    self.package = package or self.internal_package

    self.cache = {
        Plugin(self.id, str(uuid4()), self.package, tuple(topics), name=None): obj
        for topics, obj in objs
    }

add_plugin(item: Union[Collection[str], Plugin], /, plugin_obj: Any) -> None

Add an static plugin to the grappler.

Source code in grappler/grapplers/_static.py 
58
59
60
61
62
63
64
65
66
67
def add_plugin(
    self, item: Union[Collection[str], Plugin], /, plugin_obj: Any
) -> None:
    """Add an static plugin to the grappler."""
    plugin = (
        item
        if isinstance(item, Plugin)
        else Plugin(self.id, str(uuid4()), self.package, tuple(item), name=None)
    )
    self.cache[plugin] = plugin_obj

EntryPointGrappler() -> None

Bases: PluginPairGrapplerBase[metadata.EntryPoint]

A Grappler for loading objects from entry points.

Plugins are loaded from setuptools entry points installed in the Python environment. Entry point groups are mapped 1:1 to topics.

Currently, every plugin.package.platform returned from this grappler is None, even when this value is provided by underlying metadata.

Additionally, the returned plugin ids are stable across interpreter instances; this means that the plugin_id value for a given entry point definition will be the same each time this grappler iterates, between different executions of a program. This makes the grappler suitable for use with BlacklistingGrappler.

Usage:

grappler = EntryPointGrappler()
Source code in grappler/grapplers/_entry_point.py 
52
53
def __init__(self) -> None:
    self._groups = metadata.entry_points()

BouncerGrappler(inner: Optional[Grappler] = None) -> None

Bases: PluginPairGrapplerBase[None]

Restrict plugins from an inner grappler based on rules defined as predicates.

Source code in grappler/grapplers/_bouncer.py 
85
86
87
def __init__(self, inner: Optional[Grappler] = None) -> None:
    self.wrapped = inner
    self._checks = Checks(find=[], load=[])

Mode

Bases: Enum

Operating mode for checker functions.

FIND = 'find' class-attribute

For checker functions that should only be used during the grappler's scan operation. A plugin that is blocked during this operation will never be seen by grappler's client.

LOAD = 'load' class-attribute

For checker functions that should only be used during the grappler's load operation. If a plugin is blocked during this operation, an exception will be raised when attempting to load it.

checker(checker: Optional[F_Checker] = None, *, mode: Mode = Mode.BOTH) -> Union[F_Checker, Callable[[F_Checker], F_Checker]]

Register a checker function for the bouncer.

This function can be used as a decorator, or called directly:

bouncer = BouncerGrappler(...)

@bouncer.checker
def check_plugin(plugin: Plugin) -> bool
    ...

@bouncer.checker(mode=bouncer.Mode.LOAD)
def check_before_loading_only(plugin: Plugin) -> bool
    ...

def check_during_find_only(plugin: Plugin) -> bool:
    ...
bouncer.checker(check_during_find_only, mode=bouncer.Mode.FIND)
Source code in grappler/grapplers/_bouncer.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def checker(
    self, checker: Optional[F_Checker] = None, *, mode: Mode = Mode.BOTH
) -> Union[F_Checker, Callable[[F_Checker], F_Checker]]:
    """Register a checker function for the bouncer.

    This function can be used as a decorator, or called directly:

    ```python
    bouncer = BouncerGrappler(...)

    @bouncer.checker
    def check_plugin(plugin: Plugin) -> bool
        ...

    @bouncer.checker(mode=bouncer.Mode.LOAD)
    def check_before_loading_only(plugin: Plugin) -> bool
        ...

    def check_during_find_only(plugin: Plugin) -> bool:
        ...
    bouncer.checker(check_during_find_only, mode=bouncer.Mode.FIND)
    ```
    """

    def decorate(f: F_Checker) -> F_Checker:
        if mode in (self.Mode.FIND, self.Mode.BOTH):
            self._checks["find"].append(f)
        elif mode in (self.Mode.LOAD, self.Mode.BOTH):
            self._checks["load"].append(f)
        else:
            raise ValueError(f"Invalid check mode: {mode}")
        return f

    if checker is not None:
        return decorate(checker)
    else:
        return decorate

CompositeGrappler(*sources: Grappler) -> None

Bases: BasicGrappler[CompositeGrapplerIterationConfig]

Combine plugins and behaviors from multiple grapplers.

The types of grappler you can wrap with this are divided into two categories:

  • sources – A source grappler that provides plugins. Multiple of these may be provided, in which case the plugins will be chained in the order the grapplers were supplied in. These are provided to source()
  • wrappers – A special grappler that can act as a middleware, adding special behavior. Every grappler in the grappler.grapplers module which wraps a single grappler (e.g. BouncerGrappler ) may be used as a wrapper. If there is more than one wrapper, then they are executed in the order which they were provided in. They are provided to wrap()
Source code in grappler/grapplers/_composite.py 
94
95
96
97
def __init__(self, *sources: Grappler) -> None:
    self._sources = list(sources)
    self._wrappers: List[_WrappingGrappler] = []
    self._groups: Dict[str, List[Grappler]] = {}

source(source: Grappler, /, *, group: Optional[str] = None) -> CompositeGrappler

Add a source to the CompositeGrappler.

See configure_group() for the meaning of group.

Source code in grappler/grapplers/_composite.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def source(
    self, source: Grappler, /, *, group: Optional[str] = None
) -> "CompositeGrappler":
    """Add a source to the `CompositeGrappler`.

    See [`configure_group()`][grappler.grapplers.CompositeGrappler.configure_group]
    for the meaning of `group`.
    """
    self._sources.append(source)

    if group:
        self._groups.setdefault(group, []).append(source)

    return self

wrap(wrapper: _WrappingGrappler, /, *, group: Optional[str] = None) -> CompositeGrappler

Add a wrapper to the CompositeGrappler.

The grappler will be used to wrap a virtual grappler that is created from all the sources. If more than one wrapper is used, then they will be chained in the order provided.

See configure_group() for the meaning of group.

Source code in grappler/grapplers/_composite.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def wrap(
    self, wrapper: _WrappingGrappler, /, *, group: Optional[str] = None
) -> "CompositeGrappler":
    """Add a wrapper to the `CompositeGrappler`.

    The grappler will be used to wrap a virtual grappler that is
    created from all the sources. If more than one wrapper is used,
    then they will be chained in the order provided.

    See [`configure_group()`][grappler.grapplers.CompositeGrappler.configure_group]
    for the meaning of `group`.
    """
    self._wrappers.append(wrapper)

    if group:
        self._groups.setdefault(group, []).append(wrapper)

    return self

configure(target: Callable[Concatenate[Any, P_Configure], Any], *args: P_Configure.args, **kwargs: P_Configure.kwargs) -> CompositeGrappler

Call configuration functions on matching internal grapplers.

Grapplers are matched based on the host class of the configuration method passed as target. For example, is target=BouncerGrappler.checker is given, then .checker(...) will be called on every BouncerGrappler instance kept internally, with the given arguments.

This method is applied immediately, and matched against the currently registered grapplers.

Beware, the type definition for target is not fully correct. When using with a type checker, it is possible that this will allow values that will be rejected at runtime; please test thoroughly.

Source code in grappler/grapplers/_composite.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def configure(
    self,
    target: Callable[Concatenate[Any, P_Configure], Any],
    *args: P_Configure.args,
    **kwargs: P_Configure.kwargs,
) -> "CompositeGrappler":
    """
    Call configuration functions on matching internal grapplers.

    Grapplers are matched based on the host class of the configuration
    method passed as `target`. For example, is
    `target=BouncerGrappler.checker`
    is given, then `.checker(...)` will be called on every `BouncerGrappler`
    instance kept internally, with the given arguments.

    This method is applied immediately, and matched against the currently
    registered grapplers.

    Beware, the type definition for `target` is not fully correct.
    When using with a type checker, it is possible that this will allow
    values that will be rejected at runtime; please test thoroughly.
    """

    return self._configure(None, target, *args, **kwargs)

configure_group(group: str, target: Callable[Concatenate[Any, P_Configure], Any], /, *args: P_Configure.args, **kwargs: P_Configure.kwargs) -> CompositeGrappler

A version of configure() for use with named groups.

It will further narrow the matched grapplers to only those is the named group. Named groups are created by passing the group parameter to source() or wrap(). This allows to configure only grapplers with a matching group name.

Source code in grappler/grapplers/_composite.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def configure_group(
    self,
    group: str,
    target: Callable[Concatenate[Any, P_Configure], Any],
    /,
    *args: P_Configure.args,
    **kwargs: P_Configure.kwargs,
) -> "CompositeGrappler":
    """
    A version of [`configure()`][grappler.grapplers.CompositeGrappler.configure]
    for use with named groups.

    It will further narrow the matched grapplers to only those is the named
    group. Named groups are created by passing the `group` parameter to
    `source()` or `wrap()`. This allows to configure only grapplers with a
    matching group name.
    """

    return self._configure(group, target, *args, **kwargs)

BlacklistingGrappler(inner: Optional[Grappler] = None, items: Collection[Union[Plugin, Package]] = ()) -> None

Bases: BouncerGrappler, _ListGrapplerMixin

A grappler which allows to blacklist plugins or packages.

Blacklisted plugins will be blocked from iteration and loading. When a package is blacklisted, all plugins from it will be blocked.

Source code in grappler/grapplers/_list.py 
187
188
189
190
191
192
193
194
195
196
def __init__(
    self,
    inner: Optional[Grappler] = None,
    items: Collection[Union[Plugin, Package]] = (),
) -> None:
    BouncerGrappler.__init__(self, inner)
    _ListGrapplerMixin.__init__(self, items)

    # Install BouncerGrappler.checker to reject listed plugins
    self.checker(self._is_not_listed)

blacklist = _ListGrapplerMixin._add_item class-attribute

Add an item to the blacklist.

The grappler will skip any plugins that matches a blacklisted item, ensuring that they will not be loaded.

This is an instance method with the following signatures:

grappler.blacklist(plugin_or_package: Union[Plugin, Package]) -> None:
    ...

@grappler.blacklist
def get_blacklisted_plugins() -> List[Union[Plugin, Package]]:
    return [...]

These forms will blacklist plugins or entire packages by their ids. This means that no attempts will be made to match names or any other fields on the plugin on package before blocking it.

Structural matching

There are two additional signatures which perform a somewhat more structural matching (and therefore blacklisting) of plugins and packages using dictionary specifications:

grappler.blacklist({"name": "a-plugin-name", "topics": ["pytest11"]},
                    type="plugin")
grappler.blacklist({"platform": ["win32", "mac"]}, type="package")

In this form, a dictionary spec is given for matching, as well as the type argument. type must be passed as a keyword and must be either "plugin" or "package", indicating what kind of specification is being passed in.

A type of structural matching is performed on the item, in order to determine if each encountered plugin/package should be blacklisted. Every field in the dictionary is checked as an attribute of either the plugin or package, and a match is determined based on these rules:

  1. If the attribute is not present on the plugin/package, there is no match.

  2. If the field in the specification is a python collection:

    1. if the attribute is also a collection, then it matches only when the attribute is a superset of the specified collection. e.g. the "topics" field of the plugin specification above can match plugins which have plugin.topics == ("pytest11",) or plugin.topics == ("foo", "pytest11", "bar") etc.

    2. if the attribute is not a collection, then it matches only when the attribute is contained within e.g. the package specification above will blacklist only packages that either have package.platform == 'win32' or package.platform == 'mac'.

  3. In all other cases, a match only occurs when the attribute is equal to the field specification.

All fields in the specification must match a plugin/package for it to be blacklisted.

WhitelistingGrappler(inner: Optional[Grappler] = None, items: Collection[Union[Plugin, Package]] = ()) -> None

Bases: BouncerGrappler, _ListGrapplerMixin

A grappler which allows to whitelist plugins or packages

Source code in grappler/grapplers/_list.py 
278
279
280
281
282
283
284
285
286
287
def __init__(
    self,
    inner: Optional[Grappler] = None,
    items: Collection[Union[Plugin, Package]] = (),
) -> None:
    BouncerGrappler.__init__(self, inner)
    _ListGrapplerMixin.__init__(self, items)

    # Install BouncerGrappler.checker to reject non listed plugins
    self.checker(self._is_listed)

whitelist = _ListGrapplerMixin._add_item class-attribute

Add an item to the whitelist.

The grappler will skip all items unless they are in the whitelist.

This function has the same signature as BlacklistingGrappler.blacklist()