Medios-Macina/SYS/result_table_adapters.py

"""Provider registry for ResultTable API (breaking, strict API).

Providers register themselves here with an adapter and optional column factory
and selection function. Consumers (cmdlets) can look up providers by name and
obtain the columns and selection behavior for building tables and for selection
args used by subsequent cmdlets.
"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Callable, Dict, Iterable, List, Optional, Union

from SYS.result_table_api import ColumnSpec, ProviderAdapter, ResultModel, ResultTable, ensure_result_model


ColumnFactory = Callable[[List[ResultModel]], List[ColumnSpec]]
SelectionFn = Callable[[ResultModel], List[str]]


@dataclass
class Provider:
    name: str
    adapter: ProviderAdapter
    # columns can be a static list or a factory that derives columns from sample rows
    columns: Union[List[ColumnSpec], ColumnFactory]
    selection_fn: SelectionFn
    metadata: Optional[Dict[str, Any]] = None

    def get_columns(self, rows: Optional[Iterable[ResultModel]] = None) -> List[ColumnSpec]:
        if self.columns is None:
            raise ValueError(f"provider '{self.name}' must define columns")

        if callable(self.columns):
            rows_list = list(rows) if rows is not None else []
            cols = list(self.columns(rows_list))
        else:
            cols = list(self.columns)

        if not cols:
            raise ValueError(f"provider '{self.name}' produced no columns")

        return cols

    def selection_args(self, row: ResultModel) -> List[str]:
        if not callable(self.selection_fn):
            raise ValueError(f"provider '{self.name}' must define a selection function")

        sel = list(self.selection_fn(ensure_result_model(row)))
        return sel

    def build_table(self, items: Iterable[Any]) -> ResultTable:
        """Materialize adapter output into a ResultTable (strict, no legacy types)."""

        try:
            rows = [ensure_result_model(r) for r in self.adapter(items)]
        except Exception as exc:
            raise RuntimeError(f"provider '{self.name}' adapter failed") from exc

        cols = self.get_columns(rows)
        return ResultTable(provider=self.name, rows=rows, columns=cols, meta=self.metadata or {})

    def serialize_row(self, row: ResultModel) -> Dict[str, Any]:
        r = ensure_result_model(row)
        metadata = r.metadata or {}
        out: Dict[str, Any] = {
            "title": r.title,
            "path": r.path,
            "ext": r.ext,
            "size_bytes": r.size_bytes,
            "metadata": metadata,
            "source": r.source or self.name,
            "_selection_args": self.selection_args(r),
        }
        selection_action = metadata.get("_selection_action") or metadata.get("selection_action")
        if selection_action:
            out["_selection_action"] = [
                str(x) for x in selection_action if x is not None
            ]
        return out

    def serialize_rows(self, rows: Iterable[ResultModel]) -> List[Dict[str, Any]]:
        return [self.serialize_row(r) for r in rows]


_PROVIDERS: Dict[str, Provider] = {}


def register_provider(
    name: str,
    adapter: ProviderAdapter,
    *,
    columns: Union[List[ColumnSpec], ColumnFactory],
    selection_fn: SelectionFn,
    metadata: Optional[Dict[str, Any]] = None,
) -> Provider:
    name = str(name or "").strip().lower()
    if not name:
        raise ValueError("provider name required")
    if name in _PROVIDERS:
        raise ValueError(f"provider already registered: {name}")
    if columns is None:
        raise ValueError("provider registration requires columns")
    if selection_fn is None:
        raise ValueError("provider registration requires selection_fn")
    p = Provider(name=name, adapter=adapter, columns=columns, selection_fn=selection_fn, metadata=metadata)
    _PROVIDERS[name] = p
    return p


def get_provider(name: str) -> Provider:
    normalized = str(name or "").lower()
    if normalized not in _PROVIDERS:
        raise KeyError(f"provider not registered: {name}")
    return _PROVIDERS[normalized]


def list_providers() -> List[str]:
    return list(_PROVIDERS.keys())