spiele-redaktion/src/redaktionskern/plugin_loader.py

"""Plugin-Loader.

Plugins werden aus zwei Quellen geladen:
1. Installierte Pakete mit Entry-Points der Gruppe `spiele_redaktion.plugins`
   (via importlib.metadata) — für später ausgelieferte Plugin-Pakete.
2. Ein lokales `plugins/`-Verzeichnis — jeder Unterordner mit __init__.py
   ist ein Plugin (Modulattribut `plugin` = BasePlugin-Instanz).
"""
from __future__ import annotations

import importlib.metadata
import importlib.util
import sys
from importlib.metadata import entry_points
from pathlib import Path

from redaktionskern.contracts import BasePlugin

ENTRY_POINT_GROUP = "spiele_redaktion.plugins"
_MODUL_PRAEFIX = "spiele_redaktion_plugins"


class PluginError(Exception):
    """Fehler beim Laden oder Registrieren eines Plugins."""


def _als_plugin(obj: object) -> BasePlugin:
    """Akzeptiert eine Instanz oder eine Klasse (wird instanziert)."""
    if isinstance(obj, type) and issubclass(obj, BasePlugin):
        obj = obj()
    if not isinstance(obj, BasePlugin):
        raise PluginError(
            f"Erwartet wurde eine BasePlugin-Instanz, erhalten: {type(obj).__name__}"
        )
    return obj


def _aus_entry_points(group: str) -> list[BasePlugin]:
    plugins: list[BasePlugin] = []
    for entry_point in entry_points(group=group):
        try:
            plugins.append(_als_plugin(entry_point.load()))
        except PluginError:
            raise
        except Exception as exc:  # Entry-Point kann nicht geladen werden
            raise PluginError(
                f"Entry-Point '{entry_point.name}' (Gruppe {group}) "
                f"konnte nicht geladen werden: {exc}"
            ) from exc
    return plugins


def _aus_verzeichnis(verzeichnis: Path) -> list[BasePlugin]:
    plugins: list[BasePlugin] = []
    if not verzeichnis.is_dir():
        return plugins
    for eintrag in sorted(verzeichnis.iterdir()):
        if not eintrag.is_dir() or eintrag.name.startswith(("_", ".")):
            continue
        init = eintrag / "__init__.py"
        if not init.is_file():
            continue
        modulname = f"{_MODUL_PRAEFIX}.{eintrag.name}"

        # Bereits geladenes Modul wiederverwenden (idempotentes Laden):
        # Mehrere create_app()-Aufrufe im selben Prozess teilen sich sonst
        # doppelte Klassen/Tabellen auf derselben SQLAlchemy-MetaData.
        zwischengespeichert = sys.modules.get(modulname)
        if (
            zwischengespeichert is not None
            and hasattr(zwischengespeichert, "plugin")
            and getattr(zwischengespeichert.__spec__, "origin", None)
            and Path(zwischengespeichert.__spec__.origin).resolve() == init.resolve()
        ):
            plugins.append(_als_plugin(zwischengespeichert.plugin))
            continue

        spec = importlib.util.spec_from_file_location(
            modulname, init, submodule_search_locations=[str(eintrag)]
        )
        if spec is None or spec.loader is None:
            raise PluginError(f"Plugin '{eintrag.name}' kann nicht geladen werden.")
        modul = importlib.util.module_from_spec(spec)
        sys.modules[modulname] = modul
        spec.loader.exec_module(modul)
        if not hasattr(modul, "plugin"):
            raise PluginError(
                f"Plugin '{eintrag.name}' definiert kein Modulattribut 'plugin'."
            )
        plugins.append(_als_plugin(modul.plugin))
    return plugins


class PluginRegistry:
    """Verwaltet geladene Plugins; Plugin-Namen sind eindeutig."""

    def __init__(self) -> None:
        self._plugins: dict[str, BasePlugin] = {}

    def register(self, plugin: BasePlugin) -> None:
        if plugin.name in self._plugins:
            raise PluginError(
                f"Doppeltes Plugin '{plugin.name}' — Plugin-Namen müssen eindeutig sein."
            )
        self._plugins[plugin.name] = plugin

    def all(self) -> list[BasePlugin]:
        return list(self._plugins.values())

    def get(self, name: str) -> BasePlugin | None:
        return self._plugins.get(name)

    def __len__(self) -> int:
        return len(self._plugins)


def discover_plugins(
    plugins_dir: Path | None, entry_point_group: str = ENTRY_POINT_GROUP
) -> PluginRegistry:
    """Lädt Plugins aus Entry-Points und dem lokalen Verzeichnis."""
    registry = PluginRegistry()
    for plugin in _aus_entry_points(entry_point_group):
        registry.register(plugin)
    if plugins_dir is not None:
        for plugin in _aus_verzeichnis(plugins_dir):
            registry.register(plugin)
    return registry