dns-sync/dns_sync/registry.py

import importlib
import importlib.util
import os
import sys
from pathlib import Path
from types import ModuleType


_EXTERNAL_NAMESPACE = "dns_sync_external_plugins"


def _validate_plugin_name(kind: str, name: str) -> str:
    normalized = name.strip().lower()

    if (
        not normalized
        or not normalized.replace("_", "").replace("-", "").isalnum()
    ):
        raise RuntimeError(
            f"Invalid {kind} plugin name: {name!r}"
        )

    return normalized


def enabled_plugins() -> frozenset[str]:
    raw = os.getenv("ENABLED_PLUGINS", "npm")

    names = [
        item.strip().lower()
        for item in raw.split(",")
    ]

    if not names or any(not name for name in names):
        raise RuntimeError(
            "ENABLED_PLUGINS contains an empty plugin name"
        )

    validated = tuple(
        _validate_plugin_name("adapter", name)
        for name in names
    )

    if len(set(validated)) != len(validated):
        raise RuntimeError(
            "ENABLED_PLUGINS contains duplicate plugin names"
        )

    return frozenset(validated)


def _external_plugin_root() -> Path:
    return Path(
        os.getenv(
            "DNS_SYNC_PLUGIN_PATH",
            "/plugins",
        )
    )


def _ensure_external_namespace() -> None:
    if _EXTERNAL_NAMESPACE in sys.modules:
        return

    namespace = ModuleType(_EXTERNAL_NAMESPACE)
    namespace.__package__ = _EXTERNAL_NAMESPACE
    namespace.__path__ = [
        str(_external_plugin_root())
    ]
    sys.modules[_EXTERNAL_NAMESPACE] = namespace


def _load_external_adapter_module(name: str):
    module_name = name.replace("-", "_")
    package_path = (
        _external_plugin_root()
        / module_name
        / "__init__.py"
    )

    if not package_path.is_file():
        return None

    _ensure_external_namespace()

    qualified_name = (
        f"{_EXTERNAL_NAMESPACE}.{module_name}"
    )

    existing = sys.modules.get(qualified_name)
    if existing is not None:
        return existing

    spec = importlib.util.spec_from_file_location(
        qualified_name,
        package_path,
        submodule_search_locations=[
            str(package_path.parent)
        ],
    )

    if spec is None or spec.loader is None:
        raise RuntimeError(
            f"Could not load external adapter plugin "
            f"{name!r} from {package_path}"
        )

    module = importlib.util.module_from_spec(spec)
    sys.modules[qualified_name] = module

    try:
        spec.loader.exec_module(module)
    except Exception:
        sys.modules.pop(qualified_name, None)
        raise

    return module


def _load_builtin(kind: str, name: str):
    module_name = name.replace("-", "_")
    path = f"dns_sync.plugins.{kind}.{module_name}"

    try:
        return importlib.import_module(path)
    except ModuleNotFoundError as exc:
        if exc.name == path:
            return None
        raise


def _plugin_class(module, display_path: str):
    plugin_class = getattr(module, "Plugin", None)

    if plugin_class is None:
        raise RuntimeError(
            f"Plugin module {display_path} does not export Plugin"
        )

    return plugin_class


def _load(kind: str, name: str, config, *args):
    normalized = _validate_plugin_name(kind, name)

    if kind == "adapters":
        enabled = enabled_plugins()

        if normalized not in enabled:
            raise RuntimeError(
                f"Adapter plugin {normalized!r} is not enabled"
            )

    module = _load_builtin(kind, normalized)

    if module is not None:
        plugin_class = _plugin_class(
            module,
            f"dns_sync.plugins.{kind}.{normalized}",
        )
        return plugin_class(config, *args)

    if kind == "adapters":
        module = _load_external_adapter_module(
            normalized
        )

        if module is not None:
            plugin_class = _plugin_class(
                module,
                str(
                    _external_plugin_root()
                    / normalized
                    / "__init__.py"
                ),
            )
            return plugin_class(config, *args)

    raise RuntimeError(
        f"Unknown {kind[:-1]} plugin: {normalized}"
    )


def load_source(name: str, config):
    return _load("sources", name, config)


def load_provider(name: str, config):
    return _load("providers", name, config)


def load_adapter(name: str, config, endpoint):
    return _load(
        "adapters",
        name,
        config,
        endpoint,
    )