feat(harness): make tht schema columns glob/pattern-aware

At F4 (schema_linking) a model asking for a whole table family, e.g.
`fact_sost_impianto_*`, used to hit a bare "tabella non nel catalogo" and stall
without recovering. Now a pattern (containing * ? [) resolves to every matching
catalog table and returns their columns (JSON becomes an array of per-table
objects); exact names keep the original single-object contract. A non-glob miss
also suggests sibling tables sharing the leading segment, to aid recovery.

Verified live: `fact_sost_impianto_*` resolves the 20-table family on the psd
catalog. Harness suite green (811 passed).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-18 13:24:20 +02:00
co-authored by Claude Opus 4.8
parent ccfa3a18bf
commit 9fbca06f7c
2 changed files with 108 additions and 12 deletions
+49 -12
View File
@@ -393,21 +393,58 @@ def columns_cmd(
)
raise typer.Exit(code=1)
physical = PhysicalSchema.from_yaml(phys_file)
def _payload(name, tbl):
return {
"table": name,
"description": tbl.comment,
"columns": [
{"name": n, "description": col.comment, "type": col.type, "pk": col.pk}
for n, col in tbl.columns.items()
],
}
def _emit_human(p):
typer.echo(f"{p['table']}: {p['description']}")
for c in p["columns"]:
typer.echo(f" {'*' if c['pk'] else ' '} {c['name']} ({c['type']}) — {c['description']}")
# Glob-friendly: a pattern (containing * ? [) resolves to every matching catalog
# table, so a model can ask for a whole family (e.g. fact_sost_impianto_*) in one
# call instead of stalling on an unknown wildcard. Exact names keep the original
# single-object contract; JSON for a pattern is an array of per-table objects.
if any(ch in table for ch in "*?["):
from fnmatch import fnmatch
matches = sorted(n for n in physical.tables if fnmatch(n, table))
if not matches:
typer.secho(
f"ERRORE: nessuna tabella corrisponde al pattern: {table}",
fg=typer.colors.RED, err=True,
)
raise typer.Exit(code=1)
payloads = [_payload(n, physical.tables[n]) for n in matches]
if json_out:
typer.echo(_json.dumps(payloads, ensure_ascii=False))
return
for i, p in enumerate(payloads):
if i:
typer.echo("")
_emit_human(p)
return
tbl = physical.tables.get(table)
if tbl is None:
typer.secho(f"ERRORE: tabella non nel catalogo: {table}", fg=typer.colors.RED, err=True)
# Aid recovery: suggest catalog tables that share the leading segment.
prefix = table.rsplit("_", 1)[0] + "_" if "_" in table else table
hints = sorted(n for n in physical.tables if n.startswith(prefix))[:12]
msg = f"ERRORE: tabella non nel catalogo: {table}"
if hints:
msg += f" (forse: {', '.join(hints)})"
typer.secho(msg, fg=typer.colors.RED, err=True)
raise typer.Exit(code=1)
payload = {
"table": table,
"description": tbl.comment,
"columns": [
{"name": name, "description": col.comment, "type": col.type, "pk": col.pk}
for name, col in tbl.columns.items()
],
}
payload = _payload(table, tbl)
if json_out:
typer.echo(_json.dumps(payload, ensure_ascii=False))
return
typer.echo(f"{table}: {tbl.comment}")
for c in payload["columns"]:
typer.echo(f" {'*' if c['pk'] else ' '} {c['name']} ({c['type']}) — {c['description']}")
_emit_human(payload)