Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,12 @@
### Added

- **Registry: dataset mart-only**: i compose senza `clean:` non restano più fuori dal catalogo con i mart orfani. `build_clean_catalog` emette una entry dataset con metadati e `mart_refs`, senza `columns`/`location`; `registry.schema.json` rende i due campi opzionali. Così ACB/MCP vedono il dataset (`find`/`explore`) e risolvono i suoi mart (`gcs_query`).
- **`mart_name`** (#425): pipeline mode, CLI (`toolkit inspect config --mart-name`) e MCP (`toolkit_query`) selezionano la tabella mart per nome invece che per indice. `mart_index` resta per backward compat ma e' deprecato; un nome non dichiarato fallisce con l'elenco delle tabelle disponibili.
- **Registry: codelists dai support id** (#471): `build_codelists` deriva i nomi dalle entry `support:` con `type: codelist` e `id` (fonte canonica SDMX), con fallback ai `codelists/*.csv` per i repo legacy.

### Changed

- **Run MART** (#426): i parquet non piu' dichiarati in `dataset.yml` vengono rimossi dal run (per-anno e multi-year), cosi' una tabella rinominata non lascia il vecchio file a inquinare il layer tool.
- **Layering ripristinato** (PR #504): rimosse 6 inversioni di dipendenza `core → clean/domain/scout/scaffold`. `sql_validation` spostato in `domain/`, orchestration support in `domain/support.py`, `ProbePool` in `scout/pool.py`, scaffold clean.sql spostato da `raw/run.py` a `cmd_run` (comportamento invariato). Contratto condiviso `dataset_yml_slug`/`iter_workspace_dataset_ymls` in `registry/layout.py`.
- **Catalogo unico tipi fonte** (PR #505): nuovo `toolkit/core/source_types.py` come single source of truth per registry plugin, dispatch fetch e contratto AI (`contracts._SOURCE_TYPES` generato). Eliminato `_fetch_fallback` (tipo sconosciuto → `ValueError` esplicito). Tipo `script` documentato nel contratto con nota di sicurezza.

Expand Down
25 changes: 25 additions & 0 deletions tests/test_mcp_toolkit_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -752,3 +752,28 @@ def test_dataset_status_shape(tmp_path, monkeypatch):
for section in ("paths_info", "summary", "readiness", "run_stats", "info"):
assert section in result
assert result["info"]["dataset"] == dataset


def test_clean_preview_mart_name(
mcp_project_example: tuple[Path, str, int],
) -> None:
"""Issue #425: selezione della tabella mart per nome, non per indice."""
from toolkit.mcp.toolkit_client import clean_preview
from toolkit.mcp.errors import ToolkitClientError

config_path, dataset, year = mcp_project_example
root = Path(config_path).parent / "_smoke_out"
mart_dir = root / "data" / "mart" / dataset / str(year)
mart_dir.mkdir(parents=True, exist_ok=True)
_make_parquet(mart_dir / "rd_by_regione.parquet")
_make_parquet(mart_dir / "rd_by_provincia.parquet")

result = clean_preview(
str(config_path), layer="mart", mart_name="rd_by_provincia", year=year, limit=5
)
assert result["mart_name"] == "rd_by_provincia"

with pytest.raises(ToolkitClientError, match="non dichiarata"):
clean_preview(
str(config_path), layer="mart", mart_name="mart_inesistente", year=year, limit=5
)
45 changes: 45 additions & 0 deletions tests/test_pipeline_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -342,3 +342,48 @@ def test_run_mart_supports_support_placeholder(tmp_path: Path) -> None:
f"SELECT marker FROM read_parquet('{mart_output.as_posix()}')"
).fetchone() == ("ok",)
assert result["output_rows"] == 1


def test_run_mart_rimuove_output_non_dichiarati(project_example: Path, monkeypatch) -> None:
"""Issue #426: il run mart elimina i parquet non piu' dichiarati."""
_simplify_sql_project(project_example)
config_path = project_example / "dataset.yml"

monkeypatch.chdir(project_example)
cfg = load_config(config_path)
year = cfg.years[0]
logger = NoopLogger()

run_raw(
cfg.dataset,
year,
cfg.root,
cfg.raw,
logger,
base_dir=cfg.base_dir,
output_cfg=cfg.output,
clean_cfg=cfg.clean,
)
run_profile_fn(cfg, [year], logger)
run_clean(
cfg.dataset, year, cfg.root, cfg.clean, logger, base_dir=cfg.base_dir, output_cfg=cfg.output
)

mart_dir = Path(cfg.root) / "data" / "mart" / cfg.dataset / str(year)
mart_dir.mkdir(parents=True, exist_ok=True)
stale = mart_dir / "mart_vecchia.parquet"
stale.write_bytes(b"stale")

run_mart(
cfg.dataset,
year,
cfg.root,
cfg.mart,
logger,
base_dir=cfg.base_dir,
clean_cfg=cfg.clean,
output_cfg=cfg.output,
)

assert not stale.exists(), "il parquet non dichiarato deve essere rimosso"
assert (mart_dir / "rd_by_regione.parquet").exists()
46 changes: 46 additions & 0 deletions tests/test_registry_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -545,6 +545,52 @@ def test_registry_builder_imports() -> None:
# ---------------------------------------------------------------------------


@pytest.mark.contract
def test_codelists_from_support_ids(tmp_path: Path) -> None:
"""codelists derivate dai support id (SDMX), con fallback ai CSV repo."""
from toolkit.registry.builders import build_codelists

ds_dir = tmp_path / "datasets" / "demo"
ds_dir.mkdir(parents=True)
(ds_dir / "dataset.yml").write_text(
"""
root: "../../out"
schema_version: 1
dataset:
name: demo_sdmx
years: [2024]
support:
- name: freq
type: codelist
provider: sdmx
agency: ESTAT
id: FREQ
- name: unit
type: codelist
provider: sdmx
agency: ESTAT
id: unit
- name: comuni
type: dataset
config: "../support_datasets/comuni/dataset.yml"
years: [2024]
""",
encoding="utf-8",
)
cl_dir = tmp_path / "codelists"
cl_dir.mkdir()
(cl_dir / "geo.csv").write_text("code,label_en\nIT,Lombardia\n", encoding="utf-8")

layout = RepoLayout(
repo_root=tmp_path,
dataset_dirs=("datasets",),
source_repo="dataciviclab/demo",
)
payload, errors = build_codelists(layout)
assert errors == {"derive": [], "validation": []}, f"errori inattesi: {errors}"
assert payload["codelists"] == ["freq", "geo", "unit"]


@pytest.mark.contract
def test_codelists_contract(tmp_path: Path) -> None:
"""codelists espone i NOMI dei codelist del repo (contract)."""
Expand Down
6 changes: 5 additions & 1 deletion toolkit/cli/inspect/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,10 @@ def inspect_config(
year: int = typer.Option(0, "--year", "-y", help="Anno"),
sql: str | None = typer.Option(None, "--sql", help="SQL query (mode=sql)"),
limit: int = typer.Option(20, "--limit", help="Max righe (mode=preview/sql)"),
mart_index: int = typer.Option(0, "--mart-index", help="Indice tabella mart"),
mart_index: int = typer.Option(0, "--mart-index", help="Indice tabella mart (deprecato)"),
mart_name: str | None = typer.Option(
None, "--mart-name", help="Nome tabella mart (es. mart_comuni)"
),
diff: bool = typer.Option(False, "--diff", help="Schema-diff RAW tra anni"),
json_output: bool = typer.Option(False, "--json", help="Output JSON"),
):
Expand All @@ -67,6 +70,7 @@ def inspect_config(
sql=sql,
limit=limit,
mart_index=mart_index,
mart_name=mart_name,
diff=diff,
json_output=json_output,
)
Expand Down
6 changes: 5 additions & 1 deletion toolkit/cli/inspect/config_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,10 @@ def config(
year: int = typer.Option(0, "--year", "-y", help="Anno (default: ultimo configurato)"),
sql: str | None = typer.Option(None, "--sql", help="SQL query (solo mode=sql)"),
limit: int = typer.Option(20, "--limit", help="Max righe (solo mode=preview/sql)"),
mart_index: int = typer.Option(0, "--mart-index", help="Indice tabella mart (default 0)"),
mart_index: int = typer.Option(0, "--mart-index", help="Indice tabella mart (deprecato)"),
mart_name: str | None = typer.Option(
None, "--mart-name", help="Nome tabella mart (es. mart_comuni)"
),
diff: bool = typer.Option(False, "--diff", help="Schema-diff RAW tra anni"),
json_output: bool = typer.Option(False, "--json", help="Output JSON"),
) -> None:
Expand Down Expand Up @@ -122,6 +125,7 @@ def config(
limit=limit,
sql=sql,
mart_index=mart_index,
mart_name=mart_name,
)
except (ValueError, FileNotFoundError) as exc:
if json_output:
Expand Down
36 changes: 30 additions & 6 deletions toolkit/domain/layer.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,11 +68,22 @@ def _resolve_clean_path(cfg: Any, year: int) -> Path:
return Path(parquet_str)


def _resolve_mart_path(cfg: Any, year: int, mart_index: int = 0) -> Path:
def _resolve_mart_path(
cfg: Any, year: int, mart_index: int = 0, mart_name: str | None = None
) -> Path:
paths = _payload_for_year(cfg, year)
outputs = paths["paths"]["mart"].get("outputs") or []
if not outputs:
raise FileNotFoundError("Nessun output mart configurato")
if mart_name:
name = str(mart_name).strip()
for output in outputs:
if Path(output).stem == name:
return Path(output)
available = ", ".join(sorted(Path(p).stem for p in outputs))
raise ValueError(
f"Tabella mart '{name}' non dichiarata in dataset.yml: disponibili {available}"
)
if mart_index < 0 or mart_index >= len(outputs):
raise ValueError(f"Indice mart {mart_index} non valido: {len(outputs)} output disponibili")
return Path(outputs[mart_index])
Expand Down Expand Up @@ -260,6 +271,7 @@ def clean_preview(
mart_index: int = 0,
year: int | None = None,
limit: int = 10,
mart_name: str | None = None,
) -> dict[str, Any]:
"""Preview dati da un parquet clean o mart."""
from toolkit.domain.path_resolver import payload_for_year as _payload_for_year
Expand All @@ -272,7 +284,7 @@ def clean_preview(
if layer == "clean":
parquet_path = _resolve_clean_path(cfg, year)
elif layer == "mart":
parquet_path = _resolve_mart_path(cfg, year, mart_index)
parquet_path = _resolve_mart_path(cfg, year, mart_index, mart_name)
else:
raise ValueError(f"layer deve essere 'clean' o 'mart', non '{layer}'")

Expand Down Expand Up @@ -394,6 +406,7 @@ def layer_sql(
limit: int = 20,
sql: str | None = None,
mart_index: int = 0,
mart_name: str | None = None,
table: str | None = None,
dry_run: bool = False,
) -> dict[str, Any]:
Expand All @@ -411,7 +424,9 @@ def layer_sql(
limit: Max righe.
sql: Query SQL. I dati sono disponibili come tabella ``data``
(singolo dataset) o CTE col nome slug (multi dataset).
mart_index: Indice tabella mart (solo pipeline mode).
mart_index: Indice tabella mart (solo pipeline mode, deprecato:
usare ``mart_name``).
mart_name: Nome tabella mart (es ``"mart_comuni"``).
dry_run: Se True, valida lo scope SQL e fa EXPLAIN senza eseguire
(stesso contratto del dry_run di clean-query).

Expand Down Expand Up @@ -492,7 +507,7 @@ def layer_sql(
if layer == "clean":
parquet_path = _resolve_clean_path(cfg, year)
elif layer == "mart":
parquet_path = _resolve_mart_path(cfg, year, mart_index)
parquet_path = _resolve_mart_path(cfg, year, mart_index, mart_name)
else:
raise ValueError(f"layer deve essere 'raw', 'clean' o 'mart', non '{layer}'")

Expand Down Expand Up @@ -571,6 +586,7 @@ def layer_query(
limit: int = 20,
sql: str | None = None,
mart_index: int = 0,
mart_name: str | None = None,
table: str | None = None,
dry_run: bool = False,
) -> dict[str, Any]:
Expand All @@ -584,7 +600,9 @@ def layer_query(
year: Anno filtro.
limit: Max righe in preview/sql.
sql: Query SQL per mode=sql.
mart_index: Indice tabella mart (solo pipeline mode).
mart_index: Indice tabella mart (solo pipeline mode, deprecato:
usare ``mart_name``).
mart_name: Nome tabella mart (es ``"mart_comuni"``).
dry_run: Se True e mode=sql, valida lo scope + EXPLAIN senza eseguire.

Raises:
Expand Down Expand Up @@ -644,7 +662,12 @@ def layer_query(
if safe_layer == "raw":
return raw_preview(config_path, year=year, limit=limit)
return clean_preview(
config_path, layer=safe_layer, mart_index=mart_index, year=year, limit=limit
config_path,
layer=safe_layer,
mart_index=mart_index,
mart_name=mart_name,
year=year,
limit=limit,
)
if safe_mode == "sql":
return layer_sql(
Expand All @@ -655,6 +678,7 @@ def layer_query(
limit=limit,
sql=sql,
mart_index=mart_index,
mart_name=mart_name,
)

raise RuntimeError(f"mode non gestito: {safe_mode}")
29 changes: 29 additions & 0 deletions toolkit/mart/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,26 @@
_CLEAN_INPUT_TOKEN_RE = re.compile(rf"\b{CLEAN_INPUT_VIEW}\b", re.IGNORECASE)


def _cleanup_stale_mart_outputs(mart_dir: Path, declared: set[str], logger) -> list[Path]:
"""Rimuove i parquet mart non piu' dichiarati in dataset.yml (issue #426).

Le tabelle rinominate lasciavano il vecchio parquet su disco: il layer
tool poteva leggere quello sbagliato (ordine alfabetico). La pulizia gira
dopo la scrittura dei nuovi output, cosi' un run fallito non cancella
nulla.
"""
removed: list[Path] = []
if not declared or not mart_dir.is_dir():
return removed
for path in sorted(mart_dir.glob("*.parquet")):
if path.stem in declared:
continue
path.unlink()
removed.append(path)
logger.warning("MART: rimosso output non dichiarato %s", short_path(path))
return removed


def _validate_multi_year_tables(
mart_dir: Path,
mart_cfg: dict[str, Any],
Expand Down Expand Up @@ -195,6 +215,11 @@ def run_mart_multi_year(
# le table_rules/required_tables delle tabelle con years vengono
# applicate qui, dopo la produzione. Prima questa validazione non
# esisteva: le regole multi-year non venivano mai applicate.
_cleanup_stale_mart_outputs(
multi_year_dir,
{str(t.get("name")) for t in multi_year_tables if t.get("name")},
logger,
)
validation_result = _validate_multi_year_tables(
multi_year_dir,
mart_cfg,
Expand Down Expand Up @@ -536,6 +561,10 @@ def run_mart(
executed.extend(hierarchy_executed)
total_rows += hierarchy_rows

declared_names = {str(t.get("name")) for t in single_year_tables if t.get("name")}
declared_names.update(str(e["name"]) for e in executed if e.get("name"))
_cleanup_stale_mart_outputs(mart_dir, declared_names, logger)

outputs = [file_record(p) for p in written]
metadata_payload: dict[str, Any] = {
"layer": "mart",
Expand Down
6 changes: 5 additions & 1 deletion toolkit/mcp/aggregate_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ def layer_query(
limit: int = 20,
sql: str | None = None,
mart_index: int = 0,
mart_name: str | None = None,
table: str | None = None,
dry_run: bool = False,
) -> dict[str, Any]:
Expand All @@ -70,7 +71,9 @@ def layer_query(
year: Anno filtro.
limit: Max righe.
sql: Query SQL per mode=sql.
mart_index: Indice tabella mart (solo pipeline mode).
mart_index: Indice tabella mart (solo pipeline mode, deprecato:
usare ``mart_name``).
mart_name: Nome tabella mart (es ``"mart_top_sa"``).
table: Nome tabella mart (es ``"mart_top_sa"``).
dry_run: Se True (mode=sql), valida lo scope SQL e fa EXPLAIN senza eseguire.

Expand Down Expand Up @@ -98,6 +101,7 @@ def layer_query(
limit=limit,
sql=sql,
mart_index=mart_index,
mart_name=mart_name,
table=table,
dry_run=dry_run,
)
Expand Down
2 changes: 2 additions & 0 deletions toolkit/mcp/schema_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -313,6 +313,7 @@ def clean_preview(
mart_index: int = 0,
year: int | None = None,
limit: int = 10,
mart_name: str | None = None,
) -> dict[str, Any]:
"""Preview dati da un parquet clean o mart.

Expand All @@ -325,6 +326,7 @@ def clean_preview(
str(_safe_path(config_path)),
layer=layer,
mart_index=mart_index,
mart_name=mart_name,
year=year,
limit=limit,
)
Expand Down
Loading
Loading