From 2464479a0e3994f75fef2ae6147a406082b09114 Mon Sep 17 00:00:00 2001 From: Andreacnt Date: Wed, 21 Oct 2026 11:31:57 +0200 Subject: [PATCH] fix: mart_name, pulizia mart orfani e codelists da support id (#425, #426, #471) --- CHANGELOG.md | 3 ++ tests/test_mcp_toolkit_client.py | 25 ++++++++++++++++ tests/test_pipeline_integration.py | 45 +++++++++++++++++++++++++++++ tests/test_registry_builder.py | 46 ++++++++++++++++++++++++++++++ toolkit/cli/inspect/__init__.py | 6 +++- toolkit/cli/inspect/config_ops.py | 6 +++- toolkit/domain/layer.py | 36 +++++++++++++++++++---- toolkit/mart/run.py | 29 +++++++++++++++++++ toolkit/mcp/aggregate_ops.py | 6 +++- toolkit/mcp/schema_ops.py | 2 ++ toolkit/mcp/server.py | 4 ++- toolkit/registry/builders.py | 45 ++++++++++++++++++++--------- 12 files changed, 230 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3632a96..34dd909 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/tests/test_mcp_toolkit_client.py b/tests/test_mcp_toolkit_client.py index 267e0d1..14846d3 100644 --- a/tests/test_mcp_toolkit_client.py +++ b/tests/test_mcp_toolkit_client.py @@ -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 + ) diff --git a/tests/test_pipeline_integration.py b/tests/test_pipeline_integration.py index 097efa1..3fbb523 100644 --- a/tests/test_pipeline_integration.py +++ b/tests/test_pipeline_integration.py @@ -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() diff --git a/tests/test_registry_builder.py b/tests/test_registry_builder.py index f34cc90..bf4780e 100644 --- a/tests/test_registry_builder.py +++ b/tests/test_registry_builder.py @@ -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).""" diff --git a/toolkit/cli/inspect/__init__.py b/toolkit/cli/inspect/__init__.py index 53722b5..368456a 100644 --- a/toolkit/cli/inspect/__init__.py +++ b/toolkit/cli/inspect/__init__.py @@ -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"), ): @@ -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, ) diff --git a/toolkit/cli/inspect/config_ops.py b/toolkit/cli/inspect/config_ops.py index e65384d..40088d0 100644 --- a/toolkit/cli/inspect/config_ops.py +++ b/toolkit/cli/inspect/config_ops.py @@ -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: @@ -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: diff --git a/toolkit/domain/layer.py b/toolkit/domain/layer.py index 4d8fae5..d762599 100644 --- a/toolkit/domain/layer.py +++ b/toolkit/domain/layer.py @@ -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]) @@ -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 @@ -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}'") @@ -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]: @@ -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). @@ -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}'") @@ -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]: @@ -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: @@ -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( @@ -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}") diff --git a/toolkit/mart/run.py b/toolkit/mart/run.py index b3c1d1b..b3600c7 100644 --- a/toolkit/mart/run.py +++ b/toolkit/mart/run.py @@ -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], @@ -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, @@ -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", diff --git a/toolkit/mcp/aggregate_ops.py b/toolkit/mcp/aggregate_ops.py index a860470..bc7335d 100644 --- a/toolkit/mcp/aggregate_ops.py +++ b/toolkit/mcp/aggregate_ops.py @@ -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]: @@ -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. @@ -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, ) diff --git a/toolkit/mcp/schema_ops.py b/toolkit/mcp/schema_ops.py index 1bec32d..d5053fd 100644 --- a/toolkit/mcp/schema_ops.py +++ b/toolkit/mcp/schema_ops.py @@ -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. @@ -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, ) diff --git a/toolkit/mcp/server.py b/toolkit/mcp/server.py index 0410495..817c40b 100644 --- a/toolkit/mcp/server.py +++ b/toolkit/mcp/server.py @@ -188,7 +188,7 @@ def toolkit_dataset( "Query dati: SQL su dataset (raw/clean/mart) o preview URL CSV/TSV.\n\n" "Actions:\n" "- run: esegui SQL su uno o piu' dataset\n" - " Params: datasets (list[str]), sql, layer, mode, year, limit, dry_run, config_path, mart_index, table\n" + " Params: datasets (list[str]), sql, layer, mode, year, limit, dry_run, config_path, mart_index, mart_name, table\n" " Catalog mode: datasets=[slug1, slug2], sql usa gli slug come tabelle\n" " Pipeline mode: config_path, sql usa 'data' come tabella\n" "- preview: preview remoto CSV/TSV (params: url, known_encoding, known_delim, known_decimal, known_skip)" @@ -207,6 +207,7 @@ def toolkit_query( dry_run: bool = False, config_path: str | None = None, mart_index: int = 0, + mart_name: str | None = None, table: str | None = None, # preview params url: str | None = None, @@ -234,6 +235,7 @@ def toolkit_query( limit=limit, sql=sql, mart_index=mart_index, + mart_name=mart_name, table=table, dry_run=dry_run, ) diff --git a/toolkit/registry/builders.py b/toolkit/registry/builders.py index 262b938..9c42105 100644 --- a/toolkit/registry/builders.py +++ b/toolkit/registry/builders.py @@ -403,38 +403,57 @@ def _years_label(years: Iterable[int]) -> str: # --------------------------------------------------------------------------- +def _support_codelist_ids(layout: RepoLayout) -> list[str]: + """Id delle codelist dichiarate nei ``support:`` dei dataset.yml (issue #471). + + Fonte canonica: ``support:`` con ``type: codelist`` e ``id`` (id SDMX, + es. ``FREQ``, ``UNIT``). Normalizzati a lowercase e deduplicati. + """ + ids: set[str] = set() + for manifest in iter_manifests(layout): + for entry in getattr(manifest.cfg, "support", None) or []: + if not isinstance(entry, dict): + continue + if entry.get("type") not in (None, "codelist"): + continue + cid = entry.get("id") + if isinstance(cid, str) and cid.strip(): + ids.add(cid.strip().lower()) + return sorted(ids) + + def build_codelists( layout: RepoLayout, codelists_dir: Path | None = None, ) -> tuple[dict[str, Any], dict[str, list[str]]]: """Costruisce la sezione codelists del registry: i NOMI dei codelist. - Legge ``{repo_root}/codelists/*.csv`` (convenzione eurostat: geo.csv, - units.csv, freq.csv, ...) e pubblica solo i nomi come lista. Il contenuto - (le righe) NON è embeddato: è letto dal CSV del repo dal consumer on-demand - (pattern eurostat-mcp). Embeddare le righe rendeva il registry enorme + I nomi derivano dai ``support:`` con ``type: codelist`` e ``id`` dei + dataset.yml (fonte canonica, es. id SDMX) e, per compatibilita' con i + repo legacy, anche dai CSV in ``{repo_root}/codelists/*.csv`` (convenzione + eurostat: geo.csv, units.csv, freq.csv, ...). Il contenuto (le righe) NON + e' embeddato: e' letto dal CSV del repo dal consumer on-demand (pattern + eurostat-mcp). Embeddare le righe rendeva il registry enorme (es. eurostat: codelists = 60% del file) senza che alcun consumer leggesse i valori dal registry. Returns: Tuple (payload, errors) con errors = {"derive": [], "validation": [...]}. - Se la dir codelists non esiste, ritorna una lista vuota senza errori - (codelists è opzionale). + + Se non ci sono ne' support id ne' dir codelists, ritorna una lista + vuota senza errori (codelists e' opzionale). """ codelists_dir = codelists_dir or (layout.repo_root / "codelists") - if not codelists_dir.is_dir(): - return ( - {"schema_version": 1, "source_repo": layout.source_repo, "codelists": []}, - {"derive": [], "validation": []}, - ) + names = set(_support_codelist_ids(layout)) + if codelists_dir.is_dir(): + names.update(p.stem for p in codelists_dir.glob("*.csv")) - names = sorted(p.stem for p in codelists_dir.glob("*.csv")) payload: dict[str, Any] = { "schema_version": 1, "source_repo": layout.source_repo, "updated_at": str(datetime.now(UTC).date()), - "codelists": names, + "codelists": sorted(names), } return payload, {"derive": [], "validation": []}