Skip to content

runtime.plugin

EntryPointsPluginLoader

Bases: PluginLoader

Loads plugins from entry point groups

https://packaging.python.org/en/latest/specifications/entry-points/

Source code in /home/runner/work/arc/arc/arc/runtime/plugin.py 
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
class EntryPointsPluginLoader(PluginLoader):
    """Loads plugins from entry point groups

    https://packaging.python.org/en/latest/specifications/entry-points/
    """

    def __init__(
        self,
        manager: PluginManager,
        locations: t.Iterable[str],
        filter: str,
    ) -> None:
        self.filter = filter
        super().__init__(manager, locations)

    def load(self) -> None:
        for entry_point in self.__get_entry_points(self.locations):
            plugin = entry_point.load()
            self.manager.register(entry_point.name, plugin)

    def __get_entry_points(
        self, locations: t.Iterable[str]
    ) -> t.Iterator[metadata.EntryPoint]:
        for location in locations:
            for entry_point in metadata.entry_points(**{self.filter: location}):
                yield entry_point  # type: ignore

PathPluginLoader

Bases: PluginLoader

Loads plugins from a a set of file paths

Source code in /home/runner/work/arc/arc/arc/runtime/plugin.py 
 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
class PathPluginLoader(PluginLoader):
    """Loads plugins from a a set of file paths"""

    def load(self) -> None:
        for path in self.__get_paths(self.locations):
            plugin = self.__load_plugin(path)

            if plugin:
                self.manager.register(str(path), plugin)

    def __get_paths(self, paths: t.Iterable[str]) -> t.Iterator[Path]:
        for filepath in paths:
            path = self.path(filepath)
            if not path:
                continue

            if path.name.startswith("__"):
                continue

            if path.is_dir():
                yield from self.__get_paths(path.iterdir())  # type: ignore
            else:
                yield path

    def __load_plugin(self, path: Path) -> Plugin | None:
        sys.path.append(str(path.parent))
        module = import_module(path.stem)
        return getattr(module, "plugin", None)

    @staticmethod
    def path(filepath: str) -> t.Optional[Path]:
        path = Path(filepath)
        path = path.expanduser().resolve()
        if path.exists():
            return path
        return None