"""ResultTable API types and small helpers (breaking: no legacy compatibility). This module provides the canonical dataclasses and protocols that providers and renderers must use. It intentionally refuses to accept legacy dicts/strings/objs — adapters must produce `ResultModel` instances. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Callable, Dict, Iterable, List, Optional, Protocol @dataclass(frozen=True) class ResultModel: """Canonical result model that providers must produce. Fields: - title: human-friendly title (required) - path: optional filesystem path/URL - ext: file extension (without dot), e.g. "pdf", "mp3" - size_bytes: optional size in bytes - media_kind: one of 'video','audio','image','doc', etc. - metadata: arbitrary provider-specific metadata - source: provider name string """ title: str path: Optional[str] = None ext: Optional[str] = None size_bytes: Optional[int] = None media_kind: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) source: Optional[str] = None @dataclass(frozen=True) class ResultTable: """Concrete, provider-owned table of rows/columns. This is intentionally minimal: it only stores rows, column specs, and optional metadata used by renderers. It does not auto-normalize legacy objects or infer columns. """ provider: str rows: List[ResultModel] columns: List[ColumnSpec] meta: Dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: if not str(self.provider or "").strip(): raise ValueError("provider required for ResultTable") object.__setattr__(self, "rows", [ensure_result_model(r) for r in self.rows]) if not self.columns: raise ValueError("columns are required for ResultTable") object.__setattr__(self, "columns", list(self.columns)) object.__setattr__(self, "meta", dict(self.meta or {})) def serialize_row(self, row: ResultModel, selection: Optional[List[str]] = None) -> Dict[str, Any]: """Convert a row into pipeline-friendly dict (with selection args). Selection args must be precomputed by the provider; this method only copies them into the serialized dict. """ r = ensure_result_model(row) return { "title": r.title, "path": r.path, "ext": r.ext, "size_bytes": r.size_bytes, "metadata": r.metadata or {}, "source": r.source or self.provider, "_selection_args": list(selection or []), } @dataclass(frozen=True) class ColumnSpec: """Specification for a column that renderers will use. extractor: callable that accepts a ResultModel and returns the cell value. format_fn: optional callable used to format the extracted value to string. """ name: str header: str extractor: Callable[[ResultModel], Any] format_fn: Optional[Callable[[Any], str]] = None width: Optional[int] = None sortable: bool = False ProviderAdapter = Callable[[Iterable[Any]], Iterable[ResultModel]] """Type for provider adapters. Adapters must accept provider-specific sequence/iterable and yield `ResultModel` instances only. Anything else is an error (no implicit normalization). """ class Renderer(Protocol): """Renderer protocol. Implementations should accept rows and columns and return a renderable or side-effect (e.g., print to terminal). Keep implementations deterministic and side-effect-free when possible (return a Rich renderable object). """ def render(self, rows: Iterable[ResultModel], columns: Iterable[ColumnSpec], meta: Optional[Dict[str, Any]] = None) -> Any: # pragma: no cover - interface ... # Small helper enforcing strict API usage def ensure_result_model(obj: Any) -> ResultModel: """Ensure `obj` is a `ResultModel` instance, else raise TypeError. This makes the API intentionally strict: providers must construct ResultModel objects. """ if isinstance(obj, ResultModel): return obj raise TypeError("ResultModel required; providers must yield ResultModel instances") # Convenience column spec generators def title_column() -> ColumnSpec: return ColumnSpec("title", "Title", lambda r: r.title) def ext_column() -> ColumnSpec: return ColumnSpec("ext", "Ext", lambda r: r.ext or "") # Helper to build a ColumnSpec that extracts a metadata key from ResultModel def metadata_column(key: str, header: Optional[str] = None, format_fn: Optional[Callable[[Any], str]] = None) -> ColumnSpec: hdr = header or str(key).replace("_", " ").title() return ColumnSpec(name=key, header=hdr, extractor=lambda r: (r.metadata or {}).get(key), format_fn=format_fn) __all__ = [ "ResultModel", "ResultTable", "ColumnSpec", "ProviderAdapter", "Renderer", "ensure_result_model", "title_column", "ext_column", ]