Coverage for src/birdplan/plugin.py: 78%
102 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 07:38 +0000
1#
2# SPDX-License-Identifier: GPL-3.0-or-later
3#
4# Copyright (C) 2019-2025, AllWorldIT.
5#
6# This program is free software: you can redistribute it and/or modify
7# it under the terms of the GNU General Public License as published by
8# the Free Software Foundation, either version 3 of the License, or
9# (at your option) any later version.
10#
11# This program is distributed in the hope that it will be useful,
12# but WITHOUT ANY WARRANTY; without even the implied warranty of
13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14# GNU General Public License for more details.
15#
16# You should have received a copy of the GNU General Public License
17# along with this program. If not, see <https://www.gnu.org/licenses/>.
19"""Plugin handler."""
21import inspect
22import logging
23import os
24import pathlib
25import pkgutil
26from typing import Any
28__all__ = ["Plugin", "PluginCollection", "PluginMethodExceptionError", "PluginNotFoundExceptionError"]
31class PluginMethodExceptionError(RuntimeError):
32 """Plugin method exception raised when a method is called that does not exist."""
35class PluginNotFoundExceptionError(RuntimeError):
36 """Plugin not found exception raised when a plugin is referenced by name and not found."""
39class Plugin: # pylint: disable=too-few-public-methods
40 """Base plugin class, used as the parent for all plugins we define."""
42 plugin_description: str
43 plugin_order: int
45 def __init__(self) -> None:
46 """Plugin __init__ method."""
48 # Set defaults
49 self.plugin_description = type(self).__name__
50 self.plugin_order = 10
53class PluginCollection:
54 """
55 Initialize PluginCollection using a plugin base package.
57 Apon loading each plugin will be instantiated as an object.
59 Parameters
60 ----------
61 plugin_package : str
62 Source plan file to generate configuration from.
64 """
66 # The package name we will be loading plugins from
67 _plugin_packages: list[str]
68 # List of plugins we've loaded
69 _plugins: dict[str, Plugin]
70 # List of paths we've seen during processing
71 _seen_paths: list[str]
72 # Plugin statuses
73 _plugin_status: dict[str, str]
75 def __init__(self, plugin_packages: list[str]) -> None:
76 """
77 Initialize Plugincollection using a plugin base package.
79 Classes with a name ending in 'Base' will not be loaded.
81 Parameters
82 ----------
83 plugin_packages : List[str]
84 Package names to load plugins from.
86 """
88 # Setup object
89 self._plugin_packages = plugin_packages
90 self._plugins = {}
91 self._seen_paths = []
92 self._plugin_status = {}
94 # Load plugins
95 self._load_plugins()
97 def call_if_exists(self, method_name: str, args: dict[str, Any] | None = None) -> dict[str, Any]:
98 """
99 Call a plugin method, but do not raise an exception if it does not exist.
101 Parameters
102 ----------
103 method_name : str
104 Method name to call.
106 args : dict[str, Any]
107 Method argument(s).
109 Returns
110 -------
111 Dict containing the module name and its result.
113 """
115 logging.debug("Calling method '%s' if exists", method_name)
117 return self.call(method_name, args, skip_not_found=True)
119 def call(self, method_name: str, args: dict[str, Any] | None = None, skip_not_found: bool = False) -> dict[str, Any]: # noqa: FBT001,FBT002
120 """
121 Call a plugin method.
123 Parameters
124 ----------
125 method_name : str
126 Method name to call.
128 args : dict[str, Any]
129 Method argument(s).
131 skip_not_found :
132 If the method is not found return None.
134 Returns
135 -------
136 Dict containing the module name and its result.
138 """
140 # Loop with plugins, if they have overridden the method, then call it
141 results = {}
142 # Loop through plugins sorted
143 for plugin_name, plugin in sorted(self.plugins.items(), key=lambda kv: kv[1].plugin_order):
144 # Check if we're going to raise an exception or just skip
145 if not hasattr(plugin, method_name):
146 if skip_not_found:
147 logging.debug("Method '%s' does not exist in plugin '%s'", method_name, plugin_name)
148 continue
149 raise PluginMethodExceptionError(f'Plugin "{plugin_name}" has no method "{method_name}"')
150 # Save the result
151 results[plugin_name] = self.call_plugin(plugin_name, method_name, args)
153 return results
155 def get_first(self, method_name: str) -> str | None:
156 """
157 Get the first plugin method found that matches a specific method name.
159 Parameters
160 ----------
161 method_name : str
162 Method name to call.
164 Returns
165 -------
166 Any containing the result.
168 """
170 # Loop through plugins sorted
171 for plugin_name, plugin in sorted(self.plugins.items(), key=lambda kv: kv[1].plugin_order):
172 # Check if we're skipping this one if the method is not found
173 if not hasattr(plugin, method_name):
174 continue
175 # Return the first result we get
176 return plugin_name
178 return None
180 def call_first(self, method_name: str, args: dict[str, Any] | None = None) -> Any: # noqa: ANN401
181 """
182 Call the first plugin method found.
184 Parameters
185 ----------
186 method_name : str
187 Method name to call.
189 args : dict[str, Any]
190 Method argument(s).
192 Returns
193 -------
194 Any containing the result.
196 """
198 # Get first plugin which has our method
199 plugin_name = self.get_first(method_name)
201 # Make sure we got a plugin back
202 if not plugin_name:
203 raise PluginNotFoundExceptionError(f"No plugin found for method name '{method_name}'")
205 # Return the result of the method call on the first plugin
206 return self.call_plugin(plugin_name, method_name, args)
208 def call_plugin(self, plugin_name: str, method_name: str, args: dict[str, Any] | None = None) -> Any: # noqa: ANN401
209 """
210 Call a specific plugin and its method.
212 Parameters
213 ----------
214 plugin_name : str
215 Plugin to call the method in.
217 method_name : str
218 Method name to call.
220 args : dict[str, Any]
221 Method argument(s).
223 Returns
224 -------
225 Any containing the plugin call result.
227 """
229 # Check if plugin exists
230 if plugin_name not in self.plugins:
231 raise PluginNotFoundExceptionError(f'Plugin "{plugin_name}"" not found')
232 # If it does then grab it
233 plugin = self.plugins[plugin_name]
235 # Check if we're going to raise an exception or just skip
236 if not hasattr(plugin, method_name):
237 raise PluginMethodExceptionError(f'Plugin "{plugin_name}" has no method "{method_name}"')
239 # Grab the method
240 method = getattr(plugin, method_name)
242 # Call it
243 logging.debug("Calling method '%s' from plugin '%s'", method_name, plugin_name)
244 return method(args)
246 def get(self, plugin_name: str) -> Plugin:
247 """
248 Get a specific plugin object.
250 Parameters
251 ----------
252 plugin_name : str
253 Plugin to call the method in.
255 Returns
256 -------
257 Plugin object.
259 """
261 if plugin_name not in self.plugins:
262 raise PluginNotFoundExceptionError(f'Plugin "{plugin_name}" not found')
264 return self.plugins[plugin_name]
266 #
267 # Internals
268 #
270 def _load_plugins(self) -> None:
271 """Load plugins from the plugin_package we were provided."""
273 # Load plugin packages
274 for plugin_package in self._plugin_packages:
275 self._find_plugins(plugin_package)
277 def _find_plugins(self, package_name: str) -> None: # noqa: C901,PLR0912
278 """
279 Recursively search the plugin_package and retrieve all plugins.
281 Parameters
282 ----------
283 package_name : str
284 Package to load plugins from.
286 """
288 logging.debug("Finding plugins from '%s'", package_name)
290 imported_package = __import__(package_name, fromlist=["__VERSION__"])
292 # Iterate through the modules
293 for _, plugin_name, _ in pkgutil.iter_modules(imported_package.__path__, imported_package.__name__ + "."):
294 # Try import
295 try:
296 plugin_module = __import__(plugin_name, fromlist=["__VERSION__"])
297 except ModuleNotFoundError as err:
298 self._plugin_status[plugin_name] = f"cannot load module: {err}"
299 continue
301 # Grab object members
302 object_members = inspect.getmembers(plugin_module, inspect.isclass)
304 # Loop with class names
305 for _, class_name in object_members:
306 # Only add classes that are a sub class of Plugin
307 if not issubclass(class_name, Plugin) or (class_name is Plugin) or class_name.__name__.endswith("Base"):
308 continue
309 # Save plugin and record that it was loaded
310 self._plugins[plugin_name] = class_name()
311 self._plugin_status[plugin_name] = "loaded"
312 logging.debug("Plugin loaded '%s' [class=%s]", plugin_name, class_name)
314 # Look for modules in sub packages
315 all_current_paths: list[str] = []
317 if isinstance(imported_package.__path__, str):
318 all_current_paths.append(imported_package.__path__)
319 else:
320 all_current_paths.extend(imported_package.__path__)
322 # Loop with package path
323 for pkg_path in all_current_paths:
324 # Make sure its not seen in our seen_paths
325 if pkg_path in self._seen_paths:
326 continue
327 # If not add it so we don't process it again
328 self._seen_paths.append(pkg_path)
330 # Grab all the sub directories of the current package path directory
331 sub_dirs = []
332 for sub_dir in os.listdir(pkg_path): # noqa: PTH208
333 # If the subdir starts with a ., ignore it
334 if sub_dir.startswith("."):
335 continue
336 # If the subdir is __pycache__, ignore it
337 if sub_dir == "__pycache__":
338 continue
339 # If this is not a sub dir, then move onto the next one
340 if not pathlib.Path(pkg_path, sub_dir).is_dir():
341 continue
342 # Add sub-directory
343 sub_dirs.append(sub_dir)
345 # Find packages in sub directory
346 for sub_dir in sub_dirs:
347 module = f"{package_name}.{sub_dir}"
348 self._find_plugins(module)
350 @property
351 def plugins(self) -> dict[str, Plugin]:
352 """
353 Property containing the dictionary of plugins loaded.
355 Returns
356 -------
357 Dict[str, Plugin], keyed by plugin name.
359 """
361 return self._plugins
363 @property
364 def plugin_status(self) -> dict[str, str]:
365 """
366 Property containing the plugin load status.
368 Returns
369 -------
370 Dict[str, str], keyed by plugin name.
372 """
374 return self._plugin_status