leocrm/app/plugins/builtins/contracts.py

"""Central Contract Registry for inter-plugin communication.

Instead of plugins importing directly from each other's internal modules
(e.g. ``from app.plugins.builtins.kommunikation.services import send_message``),
plugins expose a **contract** module (``contracts.py``) that re-exports only
the public symbols other plugins need.

Usage pattern::

    from app.plugins.builtins.contracts import get_contract

    komm_contract = get_contract("kommunikation")
    if komm_contract:
        await komm_contract.send_message(db, ...)

This breaks the tight coupling: plugins depend on the contract surface area,
not on internal module paths.  If a plugin is absent, ``get_contract``
returns ``None`` and the caller can gracefully skip the feature.

Contracts are registered lazily on first access (import of the plugin's
``contracts`` module).  A plugin may also register itself explicitly during
``on_activate``.
"""

from __future__ import annotations

import importlib
import logging
from typing import Any, Protocol, runtime_checkable

logger = logging.getLogger(__name__)


class ContractError(Exception):
    """Raised when a contract cannot be fulfilled."""


@runtime_checkable
class PluginContract(Protocol):
    """Marker protocol for plugin contract objects.

    A contract can be any module or object that a plugin exposes via its
    ``contracts.py``.  The registry stores whatever the plugin registers.
    """

    contract_name: str


class ContractRegistry:
    """Thread-safe registry for plugin contracts.

    A contract is identified by its plugin slug (e.g. ``"kommunikation"``).
    """

    _instance: ContractRegistry | None = None

    def __new__(cls) -> ContractRegistry:
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._contracts: dict[str, Any] = {}
            cls._instance._loaded: set[str] = set()
        return cls._instance

    # ─── registration ───

    def register(self, plugin_name: str, contract: Any) -> None:
        """Register or replace a contract for a plugin."""
        self._contracts[plugin_name] = contract
        self._loaded.add(plugin_name)
        logger.debug("Contract registered for plugin '%s'", plugin_name)

    def unregister(self, plugin_name: str) -> None:
        """Remove a contract (e.g. when the plugin is deactivated)."""
        self._contracts.pop(plugin_name, None)
        self._loaded.discard(plugin_name)

    # ─── lookup ───

    def get_contract(self, plugin_name: str) -> Any | None:
        """Return the contract for *plugin_name* or ``None``.

        On first access the registry attempts to lazy-load the plugin's
        ``contracts`` module, which will register itself on import.
        """
        if plugin_name in self._contracts:
            return self._contracts[plugin_name]

        if plugin_name not in self._loaded:
            self._try_lazy_load(plugin_name)

        return self._contracts.get(plugin_name)

    def require_contract(self, plugin_name: str) -> Any:
        """Like :meth:`get_contract` but raise if unavailable."""
        contract = self.get_contract(plugin_name)
        if contract is None:
            raise ContractError(
                f"Plugin '{plugin_name}' has no registered contract. "
                "Ensure the plugin is installed and activated."
            )
        return contract

    def list_available(self) -> list[str]:
        """Return slugs of all plugins with registered contracts."""
        return sorted(self._contracts.keys())

    # ─── internals ───

    def _try_lazy_load(self, plugin_name: str) -> None:
        """Attempt to import ``app.plugins.builtins.<plugin>.contracts``.

        If the module is already in ``sys.modules`` (e.g. after a registry
        reset in tests), reload it so the registration code re-executes.
        """
        import sys

        self._loaded.add(plugin_name)  # mark as attempted even on failure
        module_path = f"app.plugins.builtins.{plugin_name}.contracts"
        try:
            if module_path in sys.modules:
                importlib.reload(sys.modules[module_path])
            else:
                importlib.import_module(module_path)
            logger.debug("Lazy-loaded contract module '%s'", module_path)
        except ImportError:
            # Plugin not installed or has no contracts module — fine.
            logger.debug("No contract module for '%s'", plugin_name)
        except Exception:
            logger.exception("Failed to load contract module '%s'", module_path)

    def _reset_for_testing(self) -> None:
        """Clear all state — for unit tests only."""
        self._contracts.clear()
        self._loaded.clear()


# ─── module-level helpers ───

def get_contract_registry() -> ContractRegistry:
    """Return the global :class:`ContractRegistry` singleton."""
    return ContractRegistry()


def get_contract(plugin_name: str) -> Any | None:
    """Convenience wrapper: ``get_contract_registry().get_contract(name)``."""
    return get_contract_registry().get_contract(plugin_name)


def require_contract(plugin_name: str) -> Any:
    """Convenience wrapper that raises if the contract is missing."""
    return get_contract_registry().require_contract(plugin_name)


def reset_contract_registry_for_testing() -> ContractRegistry:
    """Return a fresh singleton — for unit tests only."""
    reg = get_contract_registry()
    reg._reset_for_testing()
    return reg