marcopan
9b4f6b9804
WS1 of review-gates-v2: gives the JS gate (WS2) deterministic data to build the cte_result v2 payload. - CteTestRecord gains optional preview_rows (JSON-coerced, truncated cells); test_cmd populates it from the bounded result rows. - New read-only `tht cte info <name> --session <id> [--json]`: plan index/total, persisted .sql, cte_plan_doc.json entry (if any), last CteTestRecord. Exits 1 with a clean stderr message on missing session/plan/name/sql. - `tht cte plan --doc -` validates a chain-doc JSON (ctes[].name must match --name, same order) and writes it to cte_plan_doc.json; cte_plan.json stays a plain list[str] (load-bearing for tht.phase.next_cte). --doc is optional.
117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
import sqlglot
|
|
from pydantic import BaseModel
|
|
from sqlglot import exp
|
|
|
|
CTE_TESTS_FILE = "cte_tests.json"
|
|
PROBE = "\nSELECT * FROM __tht_probe__"
|
|
|
|
|
|
class CteError(Exception):
|
|
pass
|
|
|
|
|
|
class CteTestRecord(BaseModel):
|
|
name: str
|
|
ts: datetime
|
|
sql_hash: str
|
|
status: Literal["ok", "error"]
|
|
columns: list[str] = []
|
|
row_sample: int = 0
|
|
execution_ms: int = 0
|
|
warnings: list[str] = []
|
|
error: str | None = None
|
|
preview_rows: list[list[str | int | float | bool | None]] | None = None
|
|
|
|
|
|
_MAX_CELL_CHARS = 200
|
|
|
|
|
|
def _jsonable(v):
|
|
"""Coerce a raw DB cell to a JSON-native value for persistence in cte_tests.json.
|
|
|
|
None/bool/int/float pass through unchanged; str (and anything else, via str())
|
|
is truncated to ~200 chars."""
|
|
if v is None or isinstance(v, (bool, int, float)):
|
|
return v
|
|
s = v if isinstance(v, str) else str(v)
|
|
return s[:_MAX_CELL_CHARS]
|
|
|
|
|
|
def has_trailing_select(cte_sql: str) -> bool:
|
|
"""True se il file, parsato da solo, e' gia' una query completa con SELECT
|
|
finale dopo il blocco WITH (violazione della regola: il file deve contenere
|
|
solo WITH ... AS (...)).
|
|
|
|
I file CTE corretti contengono solo 'WITH ... AS (...)': sqlglot.parse_one()
|
|
fallisce su di loro perche' manca la SELECT obbligatoria.
|
|
Un file con SELECT finale parsa invece correttamente come Select con un nodo With.
|
|
I file con errori di sintassi falliscono parse_one e ritornano False (no hint)."""
|
|
try:
|
|
ast = sqlglot.parse_one(cte_sql, read="postgres")
|
|
except sqlglot.errors.ParseError:
|
|
return False
|
|
return ast is not None and ast.find(exp.With) is not None
|
|
|
|
|
|
def last_cte_name(cte_sql: str) -> str:
|
|
"""Nome dell'ultimo CTE di un blocco 'WITH a AS (...), b AS (...)' senza SELECT
|
|
finale. Trucco AV-SQL: si appende una SELECT fittizia per rendere parsabile."""
|
|
if not cte_sql.lstrip().upper().startswith("WITH"):
|
|
raise CteError("il file deve contenere un blocco WITH ... AS (...)")
|
|
try:
|
|
ast = sqlglot.parse_one(cte_sql + PROBE, read="postgres")
|
|
except sqlglot.errors.ParseError as e:
|
|
raise CteError(f"CTE non parsabile: {e}") from e
|
|
with_clause = ast.find(exp.With)
|
|
if with_clause is None or not with_clause.expressions:
|
|
raise CteError("il file deve contenere un blocco WITH ... AS (...)")
|
|
return with_clause.expressions[-1].alias
|
|
|
|
|
|
def build_test_sql(cte_sql: str) -> str:
|
|
"""SQL completo per il test: il blocco WITH + SELECT * dall'ultimo CTE.
|
|
Il LIMIT lo inietta run_controlled."""
|
|
name = last_cte_name(cte_sql)
|
|
return f"{cte_sql.rstrip().rstrip(',')}\nSELECT * FROM {name}"
|
|
|
|
|
|
def _iter_json_objects(text: str):
|
|
"""Estrae gli oggetti JSON di primo livello da `text`, qualunque sia il
|
|
formato: array JSON, JSONL, o misto (array + righe JSONL appese). Salta
|
|
spazi, virgole e parentesi d'array fra un oggetto e il successivo."""
|
|
dec = json.JSONDecoder()
|
|
i, n = 0, len(text)
|
|
while i < n:
|
|
if text[i] in " \t\r\n,[]":
|
|
i += 1
|
|
continue
|
|
obj, end = dec.raw_decode(text, i)
|
|
yield obj
|
|
i = end
|
|
|
|
|
|
def load_cte_tests(session_dir: Path) -> list[CteTestRecord]:
|
|
path = session_dir / CTE_TESTS_FILE
|
|
if not path.exists():
|
|
return []
|
|
text = path.read_text()
|
|
try:
|
|
objs = list(_iter_json_objects(text))
|
|
except json.JSONDecodeError as e:
|
|
raise CteError(f"{CTE_TESTS_FILE} corrotto: {e}") from e
|
|
return [CteTestRecord.model_validate(o) for o in objs]
|
|
|
|
|
|
def append_cte_test(session_dir: Path, record: CteTestRecord) -> None:
|
|
"""Append atomico (una riga JSON per esito): scritture concorrenti sulla
|
|
stessa sessione non si sovrascrivono, a differenza del rewrite dell'array."""
|
|
path = session_dir / CTE_TESTS_FILE
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a", encoding="utf-8") as f:
|
|
f.write(record.model_dump_json() + "\n")
|