debputy.plugin.api.experimental

src/debputy/plugin/api/experimental.py
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import re
from typing import Any, cast
from collections.abc import Callable, Iterable, Iterator

from debputy.exceptions import PluginInitializationError
from debputy.plugin.api import VirtualPath
from debputy.plugin.api.impl import DebputyPluginInitializerProvider
from debputy.plugin.api.spec import (
    DebputyPluginDefinition,
    PackageProcessor,
    PackageTypeSelector,
    DebputyPluginInitializer,
)


class ExperimentalDebputyPluginDefinition(DebputyPluginDefinition):
    """Plugin definition entity for plugins that use experimental API

    The plugin definition provides a way for the plugin code to register the features it provides via
    accessors. This particular instance also provides accessors for experimental or yet-to-be-final
    APIs. These APIs generally requires the plugin to be pre-registered with `debputy`, so `debputy`
    can add `Breaks` on the plugin provider if/when the API changes.
    """

    def package_processor(
        self,
        func: PackageProcessor | None = None,
        *,
        processor_id: str | None = None,
        depends_on_processor: Iterable[str] = tuple(),
        package_type: PackageTypeSelector = "deb",
    ) -> Callable[[PackageProcessor], PackageProcessor] | PackageProcessor:
        """Provide a pre-assembly hook that can affect the contents of the binary ("deb") packages

        The provided hook will be run once per binary package to be assembled, and it can see all the content
        ("data.tar") planned to be included in the deb. It has read-write access to the data contents.

        The hook will be run unconditionally for all binary packages built. When the hook does not apply to all
        packages, it must provide its own (internal) logic for detecting whether it is relevant and reduce itself
        to a no-op if it should not apply to the current package.

        Hooks between plugins run in "some implementation defined order" and should not rely on being run before
        or after any other hook. However, the hooks can depend on other hooks from the same plugin by using
        the `depends_on_processor`.

        The hooks are only applied to packages defined in `debian/control`. Notably, the processor will
        not apply to auto-generated `-dbgsym` packages (as those are not listed explicitly in `debian/control`).

        >>> plugin_definition = define_debputy_plugin_experimental_api()
        >>> def _la_files(fs_root: VirtualPath) -> Iterator[VirtualPath]:
        ...     ...  # Implementation skipped here for brevity
        >>>
        >>> @plugin_definition.package_processor
        ... def clean_la_files(
        ...     fs_root: "VirtualPath",
        ...     _: Any,  # Not yet defined/provided
        ...     context: "PackageProcessingContext",
        ... ) -> None:
        ...     for path in _la_files(fs_root):
        ...         buffer = []
        ...         with path.open(byte_io=True) as fd:
        ...             replace_file = False
        ...             for line in fd:
        ...                 if line.startswith(b"dependency_libs"):
        ...                     replacement = re.sub(b"<real_pattern_here>", b"''", line)
        ...                     if replacement != line:
        ...                         replace_file = True
        ...                         line = replacement
        ...                 buffer.append(line)
        ...             if not replace_file:
        ...                 continue
        ...             with path.replace_fs_path_content() as fs_path, open(fs_path, "wb") as wfd:
        ...                 wfd.writelines(buffer)

        :param func: The function to be decorated/registered.
        :param processor_id: A plugin-wide unique ID for this detector. Packagers may use this ID for disabling
          the detector and accordingly the ID is part of the plugin's API toward the packager.
        :param depends_on_processor: Denote processors from the same plugin that must run before this processor.
          It is not possible to depend on processors from other plugins.
        :param package_type: Which kind of packages this metadata detector applies to.  The package type is generally
          defined by `Package-Type` field in the binary package. The default is to only run for regular `deb` packages
          and ignore `udeb` packages.
        """

        def _decorate(f: PackageProcessor) -> PackageProcessor:

            final_id = self._name2id(processor_id, f.__name__)

            def _init(api: "DebputyPluginInitializer") -> None:
                assert isinstance(api, DebputyPluginInitializerProvider)
                api.package_processor(
                    final_id,
                    f,
                    depends_on_processor=depends_on_processor,
                    package_type=package_type,
                )

            self._generic_initializers.append(_init)

            return f

        if func:
            return _decorate(func)
        return _decorate

    def initialize(self, api: "DebputyPluginInitializer") -> None:
        """Initialize the plugin from this definition

        Most plugins will not need this function as the plugin loading will call this function
        when relevant. However, a plugin can call this manually from a function-based initializer.
        This is mostly useful if the function-based initializer need to set up a few features
        that the `DebputyPluginDefinition` cannot do and the mix/match approach is not too
        distracting for plugin maintenance.

        :param api: The plugin initializer provided by the `debputy` plugin system
        """

        initializers = self._generic_initializers
        assert isinstance(api, DebputyPluginInitializerProvider)
        if not initializers:
            plugin_name = api.plugin_metadata.plugin_name
            raise PluginInitializationError(
                f"Initialization of {plugin_name}: The plugin definition was never used to register any features."
                " If you want to conditionally register features, please use an initializer functon instead."
            )

        for initializer in initializers:
            initializer(api)


def define_debputy_plugin_experimental_api() -> ExperimentalDebputyPluginDefinition:
    return ExperimentalDebputyPluginDefinition()