Source code for httk.serve.web.widgets.loader

"""Safe loading and deterministic discovery of trusted site-local widgets."""

import hashlib
import importlib.util
import sys
import threading
from contextlib import contextmanager
from pathlib import Path
from types import ModuleType
from typing import cast

from .core import FunctionWidget, Widget, WidgetRenderer




class SiteWidgetLoader:
    """Load site widgets without allowing paths to escape their directory.

    :param widgets_dir: Directory containing trusted site widget modules.
    """

    def __init__(self, widgets_dir: Path) -> None:


        self.widgets_dir = widgets_dir

        self._cache: dict[Path, Widget] = {}
        self._cache_lock = threading.Lock()
        self._sys_path_lock = threading.Lock()



    def available(self) -> list[tuple[str, str]]:
        """List valid site-local widget modules.

        :return: Sorted ``(name, source path)`` pairs.
        """
        if not self.widgets_dir.exists():
            return []
        items: list[tuple[str, str]] = []
        for path in sorted(self.widgets_dir.glob("*.py")):
            if path.name == "__init__.py":
                continue
            if self._valid_name(path.stem):
                items.append((f"site.{path.stem}", str(path)))
        return items




    def resolve(self, name: str) -> Widget | None:
        """Resolve and load one site-local widget.

        :param name: Site widget name beginning with ``site.``.
        :return: Loaded widget, or ``None`` when the name or file is absent.
        :raises ValueError: If the widget module or definition is invalid.
        """
        if not name.startswith("site."):
            return None
        local_name = name.removeprefix("site.")
        if not self._valid_name(local_name):
            return None
        path = self._resolve_path(local_name)
        if not path.exists() or not path.is_file():
            return None
        return self._load(path, name)


    def _resolve_path(self, name: str) -> Path:
        path = (self.widgets_dir / f"{name}.py").resolve(strict=False)
        try:
            path.relative_to(self.widgets_dir.resolve(strict=False))
        except ValueError as exc:
            raise ValueError(f"Widget path escapes widgets directory: {name}") from exc
        return path

    def _load(self, path: Path, name: str) -> Widget:
        cached = self._cache.get(path)
        if cached is not None:
            return cached
        with self._cache_lock:
            cached = self._cache.get(path)
            if cached is not None:
                return cached
            module = self._load_module(path)
            definition = getattr(module, "widget", None)
            if definition is None:
                render = getattr(module, "render", None)
                if not callable(render):
                    raise ValueError(f"Widget module needs callable render(context, **props): {path}")
                definition = FunctionWidget(name=name, render_function=cast(WidgetRenderer, render), source=str(path))
            if not hasattr(definition, "render") or not callable(definition.render):
                raise ValueError(f"Widget definition has no callable render(): {path}")
            declared_name = getattr(definition, "name", name)
            if declared_name in {"", None}:
                definition = FunctionWidget(
                    name=name, render_function=cast(WidgetRenderer, definition.render), source=str(path)
                )
            elif declared_name != name:
                raise ValueError(f"Site widget name must be {name!r}, not {declared_name!r}: {path}")
            widget = cast(Widget, definition)
            self._cache[path] = widget
            return widget

    def _load_module(self, path: Path) -> ModuleType:
        digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
        spec = importlib.util.spec_from_file_location(f"httk_serve_widget_{digest}", path)
        if spec is None or spec.loader is None:
            raise ImportError(f"Cannot load widget module: {path}")
        module = importlib.util.module_from_spec(spec)
        with self._import_path(path):
            spec.loader.exec_module(module)
        return module

    @contextmanager
    def _import_path(self, path: Path):
        paths = [str(self.widgets_dir.resolve()), str(path.parent.resolve())]
        with self._sys_path_lock:
            added: list[str] = []
            for item in reversed(paths):
                if item not in sys.path:
                    sys.path.insert(0, item)
                    added.append(item)
            try:
                yield
            finally:
                for item in added:
                    sys.path.remove(item)

    @staticmethod
    def _valid_name(name: str) -> bool:
        return (
            bool(name)
            and all(c.isascii() and (c.islower() or c.isdigit() or c in {"_", "-"}) for c in name)
            and name[0].islower()
        )