File: backend_plugin_manager.py

package info (click to toggle)
errbot 6.1.7+ds-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, sid
  • size: 3,712 kB
  • sloc: python: 13,831; makefile: 164; sh: 97
file content (69 lines) | stat: -rw-r--r-- 2,049 bytes parent folder | download
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
import logging
import sys
from pathlib import Path
from typing import Any, Type

from errbot.plugin_info import PluginInfo

from .utils import collect_roots

log = logging.getLogger(__name__)


class PluginNotFoundException(Exception):
    pass


def enumerate_backend_plugins(all_plugins_paths):
    plugin_places = [Path(root) for root in all_plugins_paths]
    for path in plugin_places:
        plugfiles = path.glob("**/*.plug")
        for plugfile in plugfiles:
            plugin_info = PluginInfo.load(plugfile)
            yield plugin_info


class BackendPluginManager:
    """
    This is a one shot plugin manager for Backends and Storage plugins.
    """

    def __init__(
        self,
        bot_config,
        base_module: str,
        plugin_name: str,
        base_class: Type,
        base_search_dir,
        extra_search_dirs=(),
    ):
        self._config = bot_config
        self._base_module = base_module
        self._base_class = base_class

        self.plugin_info = None
        all_plugins_paths = collect_roots((base_search_dir, extra_search_dirs))

        for potential_plugin in enumerate_backend_plugins(all_plugins_paths):
            if potential_plugin.name == plugin_name:
                self.plugin_info = potential_plugin
                return
        raise PluginNotFoundException(
            f"Could not find the plugin named {plugin_name} in {all_plugins_paths}."
        )

    def load_plugin(self) -> Any:
        plugin_path = self.plugin_info.location.parent
        if plugin_path not in sys.path:
            # Cast pathlib.Path objects to string type for compatibility with sys.path
            sys.path.append(str(plugin_path))
        plugin_classes = self.plugin_info.load_plugin_classes(
            self._base_module, self._base_class
        )
        if len(plugin_classes) != 1:
            raise PluginNotFoundException(
                f"Found more that one plugin for {self._base_class}."
            )

        _, clazz = plugin_classes[0]
        return clazz(self._config)