260 lines
7.9 KiB
Python
260 lines
7.9 KiB
Python
|
"""Example provider that uses the new `ResultTable` API.
|
||
|
|
|
||
|
|
This module demonstrates a minimal provider adapter that yields `ResultModel`
|
||
|
|
instances, a set of `ColumnSpec` definitions, and a tiny CLI-friendly renderer
|
||
|
|
(`render_table`) for demonstration.
|
||
|
|
|
||
|
|
Run this to see sample output:
|
||
|
|
python -m Provider.example_provider
|
||
|
|
|
||
|
|
Example usage (piped selector):
|
||
|
|
provider-table -provider example -sample | select -select 1 | add-file -store default
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Dict, Iterable, List
|
||
|
|
|
||
|
|
from SYS.result_table_api import ColumnSpec, ResultModel, title_column, ext_column
|
||
|
|
|
||
|
|
|
||
|
|
SAMPLE_ITEMS = [
|
||
|
|
{
|
||
|
|
"name": "Book of Awe.pdf",
|
||
|
|
"path": "sample/Book of Awe.pdf",
|
||
|
|
"ext": "pdf",
|
||
|
|
"size": 1024000,
|
||
|
|
"source": "example",
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "Song of Joy.mp3",
|
||
|
|
"path": "sample/Song of Joy.mp3",
|
||
|
|
"ext": "mp3",
|
||
|
|
"size": 5120000,
|
||
|
|
"source": "example",
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "Cover Image.jpg",
|
||
|
|
"path": "sample/Cover Image.jpg",
|
||
|
|
"ext": "jpg",
|
||
|
|
"size": 20480,
|
||
|
|
"source": "example",
|
||
|
|
},
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def adapter(items: Iterable[Dict[str, Any]]) -> Iterable[ResultModel]:
|
||
|
|
"""Convert provider-specific items into `ResultModel` instances.
|
||
|
|
|
||
|
|
This adapter enforces the strict API requirement: it yields only
|
||
|
|
`ResultModel` instances (no legacy dict objects).
|
||
|
|
"""
|
||
|
|
for it in items:
|
||
|
|
title = it.get("name") or it.get("title") or (Path(str(it.get("path"))).stem if it.get("path") else "")
|
||
|
|
yield ResultModel(
|
||
|
|
title=str(title),
|
||
|
|
path=str(it.get("path")) if it.get("path") else None,
|
||
|
|
ext=str(it.get("ext")) if it.get("ext") else None,
|
||
|
|
size_bytes=int(it.get("size")) if it.get("size") is not None else None,
|
||
|
|
metadata=dict(it),
|
||
|
|
source=str(it.get("source")) if it.get("source") else "example",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# Columns are intentionally *not* mandated. Create a factory that inspects
|
||
|
|
# sample rows and builds only columns that make sense for the provider data.
|
||
|
|
from SYS.result_table_api import metadata_column
|
||
|
|
|
||
|
|
|
||
|
|
def columns_factory(rows: List[ResultModel]) -> List[ColumnSpec]:
|
||
|
|
cols: List[ColumnSpec] = [title_column()]
|
||
|
|
|
||
|
|
# If any row has an extension, include Ext column
|
||
|
|
if any(getattr(r, "ext", None) for r in rows):
|
||
|
|
cols.append(ext_column())
|
||
|
|
|
||
|
|
# If any row has size, include Size column
|
||
|
|
if any(getattr(r, "size_bytes", None) for r in rows):
|
||
|
|
cols.append(ColumnSpec("size", "Size", lambda rr: rr.size_bytes or "", lambda v: _format_size(v)))
|
||
|
|
|
||
|
|
# Add any top-level metadata keys discovered (up to 3) as optional columns
|
||
|
|
seen_keys = []
|
||
|
|
for r in rows:
|
||
|
|
for k in (r.metadata or {}).keys():
|
||
|
|
if k in ("name", "title", "path"):
|
||
|
|
continue
|
||
|
|
if k not in seen_keys:
|
||
|
|
seen_keys.append(k)
|
||
|
|
if len(seen_keys) >= 3:
|
||
|
|
break
|
||
|
|
if len(seen_keys) >= 3:
|
||
|
|
break
|
||
|
|
|
||
|
|
for k in seen_keys:
|
||
|
|
cols.append(metadata_column(k))
|
||
|
|
|
||
|
|
return cols
|
||
|
|
|
||
|
|
|
||
|
|
# Selection function: cmdlets rely on this to build selector args when the user
|
||
|
|
# selects a row (e.g., '@3' -> run next-cmd with the returned args). Prefer
|
||
|
|
# -path if available, otherwise fall back to -title.
|
||
|
|
def selection_fn(row: ResultModel) -> List[str]:
|
||
|
|
if row.path:
|
||
|
|
return ["-path", row.path]
|
||
|
|
return ["-title", row.title]
|
||
|
|
|
||
|
|
|
||
|
|
# Register the provider with the registry so callers can discover it by name
|
||
|
|
from SYS.result_table_adapters import register_provider
|
||
|
|
register_provider(
|
||
|
|
"example",
|
||
|
|
adapter,
|
||
|
|
columns=columns_factory,
|
||
|
|
selection_fn=selection_fn,
|
||
|
|
metadata={"description": "Example provider demonstrating dynamic columns and selectors"},
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _format_size(size: Any) -> str:
|
||
|
|
try:
|
||
|
|
s = int(size)
|
||
|
|
except Exception:
|
||
|
|
return ""
|
||
|
|
if s >= 1024 ** 3:
|
||
|
|
return f"{s / (1024 ** 3):.2f} GB"
|
||
|
|
if s >= 1024 ** 2:
|
||
|
|
return f"{s / (1024 ** 2):.2f} MB"
|
||
|
|
if s >= 1024:
|
||
|
|
return f"{s / 1024:.2f} KB"
|
||
|
|
return f"{s} B"
|
||
|
|
|
||
|
|
|
||
|
|
def render_table(rows: Iterable[ResultModel], columns: List[ColumnSpec]) -> str:
|
||
|
|
"""Render a simple ASCII table of `rows` using `columns`.
|
||
|
|
|
||
|
|
This is intentionally very small and dependency-free for demonstration.
|
||
|
|
Renderers in the project should implement the `Renderer` protocol.
|
||
|
|
"""
|
||
|
|
rows = list(rows)
|
||
|
|
|
||
|
|
# Build cell matrix (strings)
|
||
|
|
matrix: List[List[str]] = []
|
||
|
|
for r in rows:
|
||
|
|
cells: List[str] = []
|
||
|
|
for col in columns:
|
||
|
|
raw = col.extractor(r)
|
||
|
|
if col.format_fn:
|
||
|
|
try:
|
||
|
|
cell = col.format_fn(raw)
|
||
|
|
except Exception:
|
||
|
|
cell = str(raw or "")
|
||
|
|
else:
|
||
|
|
cell = str(raw or "")
|
||
|
|
cells.append(cell)
|
||
|
|
matrix.append(cells)
|
||
|
|
|
||
|
|
# Compute column widths as max(header, content)
|
||
|
|
headers = [c.header for c in columns]
|
||
|
|
widths = [len(h) for h in headers]
|
||
|
|
for row_cells in matrix:
|
||
|
|
for i, cell in enumerate(row_cells):
|
||
|
|
widths[i] = max(widths[i], len(cell))
|
||
|
|
|
||
|
|
# Helper to format a row
|
||
|
|
def fmt_row(cells: List[str]) -> str:
|
||
|
|
return " | ".join(cell.ljust(widths[i]) for i, cell in enumerate(cells))
|
||
|
|
|
||
|
|
lines: List[str] = []
|
||
|
|
lines.append(fmt_row(headers))
|
||
|
|
lines.append("-+-".join("-" * w for w in widths))
|
||
|
|
for row_cells in matrix:
|
||
|
|
lines.append(fmt_row(row_cells))
|
||
|
|
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
|
||
|
|
# Rich-based renderer (returns a Rich Table renderable)
|
||
|
|
def render_table_rich(rows: Iterable[ResultModel], columns: List[ColumnSpec]):
|
||
|
|
"""Render rows as a `rich.table.Table` for terminal output.
|
||
|
|
|
||
|
|
Returns the Table object; callers may `Console.print(table)` to render.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
from rich.table import Table as RichTable
|
||
|
|
except Exception as exc: # pragma: no cover - rare if rich missing
|
||
|
|
raise RuntimeError("rich is required for rich renderer") from exc
|
||
|
|
|
||
|
|
table = RichTable(show_header=True, header_style="bold")
|
||
|
|
for col in columns:
|
||
|
|
table.add_column(col.header)
|
||
|
|
|
||
|
|
for r in rows:
|
||
|
|
cells: List[str] = []
|
||
|
|
for col in columns:
|
||
|
|
raw = col.extractor(r)
|
||
|
|
if col.format_fn:
|
||
|
|
try:
|
||
|
|
cell = col.format_fn(raw)
|
||
|
|
except Exception:
|
||
|
|
cell = str(raw or "")
|
||
|
|
else:
|
||
|
|
cell = str(raw or "")
|
||
|
|
cells.append(cell)
|
||
|
|
table.add_row(*cells)
|
||
|
|
|
||
|
|
return table
|
||
|
|
|
||
|
|
|
||
|
|
def demo() -> None:
|
||
|
|
rows = list(adapter(SAMPLE_ITEMS))
|
||
|
|
table = render_table_rich(rows, columns_factory(rows))
|
||
|
|
try:
|
||
|
|
from rich.console import Console
|
||
|
|
except Exception:
|
||
|
|
# Fall back to plain printing if rich is not available
|
||
|
|
print("Example provider output:")
|
||
|
|
print(render_table(rows, columns_factory(rows)))
|
||
|
|
return
|
||
|
|
|
||
|
|
console = Console()
|
||
|
|
console.print("Example provider output:")
|
||
|
|
console.print(table)
|
||
|
|
|
||
|
|
|
||
|
|
def demo_with_selection(idx: int = 0) -> None:
|
||
|
|
"""Demonstrate how a cmdlet would use provider registration and selection args.
|
||
|
|
|
||
|
|
- Fetch the registered provider by name
|
||
|
|
- Build rows via adapter
|
||
|
|
- Render the table
|
||
|
|
- Show the selection args for the chosen row; these are the args a cmdlet
|
||
|
|
would append when the user picks that row.
|
||
|
|
"""
|
||
|
|
from SYS.result_table_adapters import get_provider
|
||
|
|
|
||
|
|
provider = get_provider("example")
|
||
|
|
rows = list(provider.adapter(SAMPLE_ITEMS))
|
||
|
|
cols = provider.get_columns(rows)
|
||
|
|
|
||
|
|
# Render
|
||
|
|
try:
|
||
|
|
from rich.console import Console
|
||
|
|
except Exception:
|
||
|
|
print(render_table(rows, cols))
|
||
|
|
sel_args = provider.selection_args(rows[idx])
|
||
|
|
print("Selection args for row", idx, "->", sel_args)
|
||
|
|
return
|
||
|
|
|
||
|
|
console = Console()
|
||
|
|
console.print("Example provider output:")
|
||
|
|
console.print(render_table_rich(rows, cols))
|
||
|
|
|
||
|
|
# Selection args example
|
||
|
|
sel = provider.selection_args(rows[idx])
|
||
|
|
console.print("Selection args for row", idx, "->", sel)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
demo()
|