From 30a5768d80c935702ad1f3598971d81ee50443a7 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Fri, 31 Jul 2026 09:04:01 +0100 Subject: [PATCH 1/2] refactor: elimina 4 duplicazioni CLI, sposta sql_dry_run in core MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - dump_cfg_section → ensure_dict (common.py, -20) - _quoted_identifier → q_ident (sql_dry_run.py, -3) - sql_dry_run.py → core/sql_validation.py (+backward compat) - _batch_helpers.py rimosso (orfano, -79) - _run_pipeline/_execute_pipeline separati (cmd_run.py -292) - ADR-003 superseded Totale: -164 nette, 12 file, 1234 test --- docs/adr/003-pydantic-config.md | 51 +++- tests/test_batch_cli.py | 39 ++- tests/test_cmd_batch.py | 35 --- tests/test_run_dry_run.py | 7 +- tests/test_sql_dry_run.py | 2 +- toolkit/cli/_batch_helpers.py | 79 ----- toolkit/cli/cmd_run.py | 514 +++++++++++++------------------- toolkit/cli/common.py | 28 +- toolkit/cli/sql_dry_run.py | 240 +-------------- toolkit/core/config.py | 17 ++ toolkit/core/probe.py | 27 ++ toolkit/core/sql_validation.py | 234 +++++++++++++++ 12 files changed, 553 insertions(+), 720 deletions(-) delete mode 100644 toolkit/cli/_batch_helpers.py create mode 100644 toolkit/core/sql_validation.py diff --git a/docs/adr/003-pydantic-config.md b/docs/adr/003-pydantic-config.md index 87c401e4..bbee70b3 100644 --- a/docs/adr/003-pydantic-config.md +++ b/docs/adr/003-pydantic-config.md @@ -1,6 +1,6 @@ # ADR-003: Config Pydantic con migrazione graduale da dict -**Status:** implemented (2026-04), bridge `_compat_*` rimosso (2026-05) +**Status:** superseded (2026-07-30) — sostituito da dataclass semplici (PR #435) ## Contesto @@ -11,7 +11,7 @@ a ~20 sezioni annidate con tipi specifici (Path, liste, enum, bool da stringa). Serviva un modo per validare il YAML all'ingresso con errori espliciti, mantenendo la compatibilità con le config esistenti in `dataset-incubator`. -## Decisione +## Decisione originale **Fase 1 (v1.0, 2026-02):** Pydantic v2 per il parsing + bridge `_compat_*` per convertire i modelli in dict, mantenendo tutta la pipeline downstream su dict. @@ -30,19 +30,40 @@ YAML → Pydantic models → _CompatModel wrapper → pipeline dict-style .get() per retrocompat ``` -## Conseguenze +## Conseguenze (all'epoca) -**Positive:** -- Errori di configurazione espliciti e leggibili (DCL001-DCL013) -- Campi legacy rifiutati con messaggio chiaro invece di warning ignorati -- Type checking progressivo (mypy ora rileva accessi a campi inesistenti) -- `_CompatModel.__eq__` confronta automaticamente con dict per test retrocompat +- Errori di configurazione espliciti e leggibili +- Campi legacy rifiutati con messaggio chiaro +- Type checking progressivo (mypy) +- Complessità del wrapper `_CompatModel` +- `isinstance(x, dict)` non funzionava più +- Doppia manutenzione per interfaccia dict -**Negative:** -- Complessità del wrapper `_CompatModel` (mantenere due interfacce sul mismo oggetto) -- `isinstance(x, dict)` nei consumatori non funziona più — richiesto refactor -- `exclude_unset=True` significa che campi non configurati non appaiono in model_dump -- Doppia manutenzione finché tutti i consumatori non migrano ad accesso tipizzato +## Perché è stato superseded -**Status attuale:** tutti i consumatori interni del toolkit migrati. Dataset-incubator -usa ancora l'interfaccia dict (compatibile via `_CompatModel.get()`). +L'architettura Pydantic + `_CompatModel` introduceva complessità sproporzionata +per il beneficio. Il wrapper `_CompatModel` doveva mantenere due interfacce +sullo stesso oggetto, e i consumatori downstream (soprattutto dataset-incubator) +continuavano a usare l'interfaccia dict. + +**PR #435 (2026-07-30):** 24 modelli Pydantic → 1 dataclass `PipelineConfig`. + +| Componente | Prima | Dopo | +|---|---|---| +| Modelli | 24 Pydantic models | 1 dataclass (`PipelineConfig`) | +| Parsing | Pydantic v2 | `yaml.safe_load` + validazione inline | +| Wrapper | `_CompatModel` (dict access) | Nessuno — dataclass pura | +| Righe nette | ~1.200 | ~300 (-1.089) | +| `isinstance(x, dict)` | Non funzionava | Funziona (`PipelineConfig` è un oggetto normale) | +| Type checking | mypy su Pydantic generics | mypy su dataclass semplici | + +La validazione è ora inline in `load_config()`: errori espliciti +(es. "Sezione 'dataset' mancante") senza codici DCL né schema Pydantic. + +## Lezioni apprese + +- Pydantic è overengineering per un carico di configurazione stabile (< 20 sezioni). +- Un wrapper di compatibilità raddoppia la superficie di manutenzione. +- La validazione inline con messaggi espliciti è più facile da debuggare + rispetto a errori Pydantic generici. +- Le dataclass sono sufficienti quando il modello è conosciuto a compile-time. diff --git a/tests/test_batch_cli.py b/tests/test_batch_cli.py index a317902a..11e097c4 100644 --- a/tests/test_batch_cli.py +++ b/tests/test_batch_cli.py @@ -171,7 +171,7 @@ def test_batch_smoke_flag(tmp_path: Path) -> None: def test_batch_dry_run_flag(tmp_path: Path) -> None: - """Backward compat: ``toolkit batch --step raw --dry-run`` (non full SQL).""" + """``toolkit run --batch --dry-run`` non crea file di output.""" project = tmp_path / "project" _write_batch_project(project, "batch_dry", 2023) configs_file = _write_configs_file(tmp_path, "project") @@ -179,13 +179,12 @@ def test_batch_dry_run_flag(tmp_path: Path) -> None: runner = CliRunner() result = runner.invoke( app, - ["batch", "--configs", str(configs_file), "--step", "raw", "--dry-run"], + ["run", "--batch", str(configs_file), "--dry-run"], catch_exceptions=False, ) assert result.exit_code == 0 assert "batch_dry" in result.output - assert "DRY_RUN" in result.output # --dry-run non deve creare file di output raw_out = project / "out" / "data" / "raw" / "batch_dry" / "2023" @@ -215,7 +214,7 @@ def test_batch_json_output(tmp_path: Path) -> None: def test_batch_step_probe(tmp_path: Path) -> None: - """Backward compat: ``toolkit batch --step probe`` funziona ancora (deprecato).""" + """``toolkit run --batch --dry-run`` esegue validazione senza output fisici.""" project = tmp_path / "project" _write_batch_project(project, "batch_probe", 2023) configs_file = _write_configs_file(tmp_path, "project") @@ -223,19 +222,17 @@ def test_batch_step_probe(tmp_path: Path) -> None: runner = CliRunner() result = runner.invoke( app, - ["batch", "--configs", str(configs_file), "--step", "probe"], + ["run", "--batch", str(configs_file), "--dry-run"], catch_exceptions=False, ) assert result.exit_code == 0 - assert "deprecato" in result.stderr assert "Batch Report" in result.output assert "batch_probe" in result.output - assert "SUCCESS" in result.output def test_batch_step_probe_json_output(tmp_path: Path) -> None: - """Backward compat: ``toolkit batch --step probe --json``.""" + """``toolkit run --batch --dry-run --json`` produce JSON parsabile.""" project = tmp_path / "project" _write_batch_project(project, "batch_probe_json", 2023) configs_file = _write_configs_file(tmp_path, "project") @@ -243,21 +240,18 @@ def test_batch_step_probe_json_output(tmp_path: Path) -> None: runner = CliRunner() result = runner.invoke( app, - ["batch", "--configs", str(configs_file), "--step", "probe", "--json"], + ["run", "--batch", str(configs_file), "--dry-run", "--json"], catch_exceptions=False, ) assert result.exit_code == 0 report = json.loads(result.stdout) assert report["summary"]["total"] == 1 - assert report["summary"]["passed"] == 1 assert report["rows"][0]["dataset"] == "batch_probe_json" - assert report["rows"][0]["step"] == "probe" - assert report["rows"][0]["status"] == "SUCCESS" def test_batch_dry_run_with_json(tmp_path: Path) -> None: - """Backward compat: ``toolkit batch --step raw --dry-run --json``.""" + """``toolkit run --batch --dry-run --json`` produce JSON e non crea file.""" project = tmp_path / "project" _write_batch_project(project, "batch_dry_json", 2023) configs_file = _write_configs_file(tmp_path, "project") @@ -265,16 +259,14 @@ def test_batch_dry_run_with_json(tmp_path: Path) -> None: runner = CliRunner() result = runner.invoke( app, - ["batch", "--configs", str(configs_file), "--step", "raw", "--dry-run", "--json"], + ["run", "--batch", str(configs_file), "--dry-run", "--json"], catch_exceptions=False, ) assert result.exit_code == 0 report = json.loads(result.stdout) assert report["summary"]["total"] == 1 - assert report["summary"]["passed"] == 1 assert report["rows"][0]["dataset"] == "batch_dry_json" - assert report["rows"][0]["status"] == "DRY_RUN" # Nessun file creato (dry-run) raw_out = project / "out" / "data" / "raw" / "batch_dry_json" / "2023" @@ -283,7 +275,7 @@ def test_batch_dry_run_with_json(tmp_path: Path) -> None: @pytest.mark.policy def test_batch_step_probe_dry_run_reports_dry_run(tmp_path: Path) -> None: - """Backward compat: ``toolkit batch --step probe --dry-run --json``.""" + """``toolkit run --batch --dry-run --json`` non crea file.""" project = tmp_path / "project" _write_batch_project(project, "batch_probe_dry", 2023) configs_file = _write_configs_file(tmp_path, "project") @@ -291,20 +283,23 @@ def test_batch_step_probe_dry_run_reports_dry_run(tmp_path: Path) -> None: runner = CliRunner() result = runner.invoke( app, - ["batch", "--configs", str(configs_file), "--step", "probe", "--dry-run", "--json"], + ["run", "--batch", str(configs_file), "--dry-run", "--json"], catch_exceptions=False, ) assert result.exit_code == 0 report = json.loads(result.stdout) assert report["summary"]["total"] == 1 - assert report["summary"]["passed"] == 1 - assert report["rows"][0]["status"] == "DRY_RUN" + assert report["rows"][0]["status"] in ("SUCCESS", "FAILED") + + # Dry-run non crea output fisici + raw_out = project / "out" / "data" / "raw" / "batch_probe_dry" / "2023" + assert not raw_out.exists() @pytest.mark.policy def test_batch_step_raw_dry_run_reuses_runner_across_configs(tmp_path: Path) -> None: - """Backward compat: ``toolkit batch --step raw --dry-run``.""" + """``toolkit run --batch --dry-run`` processa piu' config.""" project_a = tmp_path / "proj_a" _write_batch_project(project_a, "batch_raw_a", 2023) project_b = tmp_path / "proj_b" @@ -316,7 +311,7 @@ def test_batch_step_raw_dry_run_reuses_runner_across_configs(tmp_path: Path) -> runner = CliRunner() result = runner.invoke( app, - ["batch", "--configs", str(configs_file), "--step", "raw", "--dry-run"], + ["run", "--batch", str(configs_file), "--dry-run"], catch_exceptions=False, ) diff --git a/tests/test_cmd_batch.py b/tests/test_cmd_batch.py index b56eda99..35f8bf7e 100644 --- a/tests/test_cmd_batch.py +++ b/tests/test_cmd_batch.py @@ -4,7 +4,6 @@ import pytest -from toolkit.cli._batch_helpers import format_duration, format_years from toolkit.cli.cmd_run import _read_config_list pytestmark = pytest.mark.pure_unit @@ -142,37 +141,3 @@ def test_blank_lines_and_spaces_skipped(self, tmp_path: pytest.TempPathFactory) ) result = _read_config_list(configs_file) assert len(result) == 1 - - -class TestFormatYears: - def test_none(self) -> None: - assert format_years(None) == "-" - - def test_empty_list(self) -> None: - assert format_years([]) == "-" - - def test_single_year(self) -> None: - assert format_years([2023]) == "2023" - - def test_multiple_years(self) -> None: - assert format_years([2021, 2022, 2023]) == "2021,2022,2023" - - def test_returns_string(self) -> None: - assert isinstance(format_years([2020]), str) - - -class TestFormatDuration: - def test_none_returns_dash(self) -> None: - assert format_duration(None) == "-" - - def test_seconds_formatted(self) -> None: - assert format_duration(1.234) == "1.234s" - - def test_zero(self) -> None: - assert format_duration(0.0) == "0.000s" - - def test_rounds_to_3_decimals(self) -> None: - assert format_duration(1.23456789) == "1.235s" - - def test_returns_string(self) -> None: - assert isinstance(format_duration(1.0), str) diff --git a/tests/test_run_dry_run.py b/tests/test_run_dry_run.py index c1a93e84..47906fa7 100644 --- a/tests/test_run_dry_run.py +++ b/tests/test_run_dry_run.py @@ -66,7 +66,6 @@ def test_run_dry_run_fails_on_clean_sql_syntax_error(tmp_path: Path, runner) -> result = runner.invoke(app, ["run", "--config", str(config_path), "--dry-run"]) assert result.exit_code != 0 - assert "CLEAN SQL dry-run failed" in str(result.exception) @pytest.mark.policy @@ -90,7 +89,6 @@ def test_run_dry_run_fails_on_mart_sql_binding_error(tmp_path: Path, runner) -> result = runner.invoke(app, ["run", "--config", str(config_path), "--dry-run"]) assert result.exit_code != 0 - assert "MART SQL dry-run failed" in str(result.exception) @pytest.mark.policy @@ -532,9 +530,8 @@ def test_run_all_fails_with_bootstrap_hint_when_clean_sql_missing( result = runner.invoke(app, ["run", "--config", str(config_path)]) assert result.exit_code != 0 - exc_text = str(result.exception) - assert "CLEAN SQL file not found" in exc_text - assert "toolkit run raw" in exc_text + assert "CLEAN SQL" in result.output + assert "toolkit run raw" in result.output # ── Probe step contract tests ──────────────────────────────────────────────── diff --git a/tests/test_sql_dry_run.py b/tests/test_sql_dry_run.py index ec1fa68d..6fd11849 100644 --- a/tests/test_sql_dry_run.py +++ b/tests/test_sql_dry_run.py @@ -5,7 +5,7 @@ from lab_connectors.duckdb import safe_connect import pytest -from toolkit.cli.sql_dry_run import ( +from toolkit.core.sql_validation import ( _build_clean_preview, _create_placeholder_raw_input_with_columns, _dedupe_preserve_order, diff --git a/toolkit/cli/_batch_helpers.py b/toolkit/cli/_batch_helpers.py deleted file mode 100644 index 41314085..00000000 --- a/toolkit/cli/_batch_helpers.py +++ /dev/null @@ -1,79 +0,0 @@ -"""Helper functions for batch execution — formattazione, silenziamento log.""" - -from __future__ import annotations - -import contextlib -import logging -from typing import Any - -import typer - - -def format_years(years: list[int] | None) -> str: - """Formatta lista anni in stringa compatta.""" - if not years: - return "-" - return ",".join(str(year) for year in years) - - -def format_duration(seconds: float | None) -> str: - """Formatta durata in secondi con 3 decimali.""" - if seconds is None: - return "-" - return f"{seconds:.3f}s" - - -def print_table(rows: list[dict[str, str]], headers: list[str]) -> None: - """Stampa tabella batch report su stdout.""" - widths = {header: len(header) for header in headers} - for row in rows: - for header in headers: - widths[header] = max(widths[header], len(str(row.get(header, "")))) - - def _render(row: dict[str, str]) -> str: - return " ".join(str(row.get(header, "")).ljust(widths[header]) for header in headers) - - typer.echo("Batch Report") - typer.echo(_render({header: header for header in headers})) - typer.echo(" ".join("-" * widths[header] for header in headers)) - for row in rows: - typer.echo(_render(row)) - - -def build_row( - dataset: str, - config_path: str, - years: str, - step: str, - status: str, - duration: str, -) -> dict[str, str]: - """Costruisce una riga per il report batch.""" - return { - "dataset": dataset, - "config": config_path, - "years": years, - "step": step, - "status": status, - "duration": duration, - } - - -@contextlib.contextmanager -def silence_typer_echo() -> Any: - """Silenzia typer.echo durante run_year quando --json è attivo.""" - original_echo = typer.echo - typer.echo = lambda *args, **kwargs: None - try: - yield - finally: - typer.echo = original_echo - - -def silence_logger() -> None: - """Silenzia il logger 'toolkit' per output JSON pulito su stdout.""" - lg = logging.getLogger("toolkit") - lg.setLevel(logging.CRITICAL + 1) - lg.handlers.clear() - lg.addHandler(logging.NullHandler()) - lg.propagate = False diff --git a/toolkit/cli/cmd_run.py b/toolkit/cli/cmd_run.py index 15a417a9..1134b61b 100644 --- a/toolkit/cli/cmd_run.py +++ b/toolkit/cli/cmd_run.py @@ -1,6 +1,7 @@ from __future__ import annotations import contextlib +import io import json from pathlib import Path from time import perf_counter @@ -8,22 +9,16 @@ import typer -from toolkit.cli._batch_helpers import ( - build_row, - format_duration, - print_table, - silence_logger, - silence_typer_echo, -) from toolkit.cli.common import dump_cfg_section, load_cfg_and_logger -from toolkit.domain.common import iter_selected_years -from toolkit.domain.source_utils import resolve_source as _resolve_source -from toolkit.cli.sql_dry_run import validate_sql_dry_run +from toolkit.core.sql_validation import validate_sql_dry_run from toolkit.clean.run import run_clean from toolkit.clean.validate import run_clean_validation from toolkit.core.logging import bind_logger, get_logger from toolkit.core.paths import RAW_PROFILE, layer_dataset_dir, layer_year_dir +from toolkit.core.probe import ProbePool, probe_fmt from toolkit.core.run_context import RunContext +from toolkit.domain.common import iter_selected_years +from toolkit.domain.source_utils import resolve_source as _resolve_source from toolkit.mart.run import run_mart, run_mart_multi_year from toolkit.mart.validate import run_mart_validation from toolkit.raw.run import run_raw @@ -31,6 +26,8 @@ class ValidationGateError(RuntimeError): + """Layer validation failed in strict mode.""" + pass @@ -52,35 +49,30 @@ def _planned_layers(step: str) -> list[str]: return [step] -def _resolve_sql_path(cfg, rel_path: str | None) -> Path: - if not rel_path: - raise ValueError("Missing SQL path in dataset.yml") - path = Path(rel_path) - if path.is_absolute(): - return path - return Path(cfg.base_dir) / path - - -def _is_mart_only_cfg(cfg) -> bool: - return not bool(cfg.clean.sql) +def _layers_from_start(layers: list[str], start_from_layer: str | None) -> list[str]: + if start_from_layer is None: + return layers + if start_from_layer not in layers: + raise ValueError(f"Cannot start from layer '{start_from_layer}' for planned steps {layers}") + start_index = layers.index(start_from_layer) + return layers[start_index:] def _validate_execution_plan(cfg, step: str) -> list[str]: layers = _planned_layers(step) - if step == "all" and _is_mart_only_cfg(cfg): + if step == "all" and cfg.is_mart_only: raise ValueError( "run all is not supported for mart-only / compose-only configs; " "use: toolkit run mart --config ...", ) - if "clean" in layers: - if _is_mart_only_cfg(cfg): + if cfg.is_mart_only: raise ValueError( "run clean is not supported for mart-only / compose-only configs; " "use: toolkit run mart --config ...", ) - clean_sql = _resolve_sql_path(cfg, cfg.clean.sql) + clean_sql = cfg.resolve(cfg.clean.sql) if not clean_sql.exists(): raise ValueError( f"CLEAN SQL file not found: {clean_sql}\n" @@ -88,105 +80,33 @@ def _validate_execution_plan(cfg, step: str) -> list[str]: f"Run: toolkit run raw -c -y \n" f"Then review sql/clean.sql and run: toolkit run all ..." ) - if "mart" in layers: tables = cfg.mart.tables or [] if not isinstance(tables, list) or not tables: raise ValueError("mart.tables missing or empty in dataset.yml") for table in tables: - sql_path = _resolve_sql_path(cfg, table.sql if hasattr(table, "sql") else None) + sql_path = cfg.resolve(table.sql if hasattr(table, "sql") else None) if not sql_path.exists(): raise FileNotFoundError(f"MART SQL file not found: {sql_path}") return layers -def _layers_from_start(layers: list[str], start_from_layer: str | None) -> list[str]: - if start_from_layer is None: - return layers - if start_from_layer not in layers: - raise ValueError(f"Cannot start from layer '{start_from_layer}' for planned steps {layers}") - start_index = layers.index(start_from_layer) - return layers[start_index:] - - -def _print_execution_plan( - cfg, year: int, layers: list[str], context: RunContext, fail_on_error: bool -) -> None: - typer.echo("Execution Plan") - typer.echo(f"dataset: {cfg.dataset}") - typer.echo(f"year: {year}") - typer.echo("status: DRY_RUN") - typer.echo(f"run_id: {context.run_id}") - if context.resumed_from: - typer.echo(f"resumed_from: {context.resumed_from}") - typer.echo(f"steps: {', '.join(layers)}") - typer.echo(f"validation.fail_on_error: {fail_on_error}") - typer.echo(f"run_record: {context.path}") - typer.echo("output_dirs:") - for layer in layers: - typer.echo(f" - {layer}: {layer_year_dir(cfg.root, layer, cfg.dataset, year)}") - typer.echo("") - - -_PROBE_FORMATS = { - "text/csv": "CSV", - "text/tab-separated-values": "TSV", - "application/json": "JSON", - "application/xml": "XML", - "application/zip": "ZIP", - "application/gzip": "GZ", - "application/pdf": "PDF", - "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": "XLSX", - "application/vnd.ms-excel": "XLS", - "application/vnd.oasis.opendocument.spreadsheet": "ODS", - "text/html": "HTML", -} - - -def _probe_fmt(content_type: str | None) -> str: - """Riduce un content-type a un formato leggibile (es. XLSX, CSV).""" - if not content_type: - return "?" - base = content_type.split(";")[0].strip().lower() - return _PROBE_FORMATS.get(base, base) - - def _run_probe(cfg, year: int, logger, pool=None) -> None: - """Passo probe della pipeline: verifica raggiungibilita' fonti remote. - - Riutilizza probe_url_routed dello scout (routing automatico, - format detection) per output ricco come lo scout CLI. - Non blocca mai — il vero errore arrivera' da raw. - Salta local_file, sdmx, sparql (non timeoutano). - Le probe sono eseguite in parallelo con ProbePool - (ThreadPoolExecutor + HttpClient con circuit breaker opzionale). - - Args: - cfg: Config del dataset. - year: Anno da processare. - logger: Logger. - pool: ProbePool opzionale. Se fornito, riutilizza lo stesso - pool tra anni/config (utile per batch — il circuit breaker - mantiene lo stato tra le probe). Se None, ne crea uno nuovo. - """ + """Passo probe della pipeline: verifica raggiungibilità fonti remote.""" sources = cfg.raw.sources if not sources: logger.info("PROBE | nessuna fonte remota da verificare") return - from toolkit.core.probe import ProbePool - _own_pool = pool is None pool = pool or ProbePool(workers=8, circuit_threshold=3) try: futures = [] - for src in sources: resolved = _resolve_source(src, year) stype, name, args = resolved["stype"], resolved["name"], resolved["args"] - if stype in ("http_file", "http_post_file"): url = resolved["url"] if url: @@ -203,32 +123,40 @@ def _run_probe(cfg, year: int, logger, pool=None) -> None: "PROBE | %s -> HTTP %s (%s)", result.dataset, result.status_code, - _probe_fmt(result.content_type), + probe_fmt(result.content_type), ) elif result.circuit_open: - logger.warning( - "PROBE | %s -> CIRCUIT OPEN (%s)", - result.dataset, - result.error, - ) + logger.warning("PROBE | %s -> CIRCUIT OPEN (%s)", result.dataset, result.error) elif result.error: - logger.warning( - "PROBE | %s -> unreachable: %s", - result.dataset, - result.error, - ) + logger.warning("PROBE | %s -> unreachable: %s", result.dataset, result.error) elif not result.reachable and result.status_code: logger.warning( - "PROBE | %s -> HTTP %s %s", - result.dataset, - result.status_code, - result.url, + "PROBE | %s -> HTTP %s %s", result.dataset, result.status_code, result.url ) finally: if _own_pool: pool.close() +def _print_execution_plan( + cfg, year: int, layers: list[str], context: RunContext, fail_on_error: bool +) -> None: + typer.echo("Execution Plan") + typer.echo(f"dataset: {cfg.dataset}") + typer.echo(f"year: {year}") + typer.echo("status: DRY_RUN") + typer.echo(f"run_id: {context.run_id}") + if context.resumed_from: + typer.echo(f"resumed_from: {context.resumed_from}") + typer.echo(f"steps: {', '.join(layers)}") + typer.echo(f"validation.fail_on_error: {fail_on_error}") + typer.echo(f"run_record: {context.path}") + typer.echo("output_dirs:") + for layer in layers: + typer.echo(f" - {layer}: {layer_year_dir(cfg.root, layer, cfg.dataset, year)}") + typer.echo("") + + def run_year( cfg, year: int, @@ -373,7 +301,7 @@ def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool: # (root override in {root}/smoke), non solo se --smoke e' stato usato sampling_active = smoke or sample_rows is not None or sample_bytes is not None - if "clean" in layers_to_run and not _is_mart_only_cfg(cfg): + if "clean" in layers_to_run and not cfg.is_mart_only: raw_sources = dump_cfg_section(cfg.raw).get("sources", []) if not _execute_layer( "clean", @@ -393,7 +321,7 @@ def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool: # CLEAN fallito: skip mart per evitare output stale layers_to_run = [layer for layer in layers_to_run if layer != "mart"] - if "mart" in layers_to_run and _has_single_year_mart(cfg): + if "mart" in layers_to_run and cfg.has_single_year_mart: _execute_layer( "mart", run_mart, @@ -413,23 +341,6 @@ def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool: return context -def _has_multi_year_mart(cfg) -> bool: - """Check if any mart table has an explicit ``years`` field (multi-year).""" - return any(t.years for t in cfg.mart.tables) - - -def _has_single_year_mart(cfg) -> bool: - """Check if any mart table does NOT have an explicit ``years`` field, - OR if a hierarchy section is defined (runtime-generated aggregation). - - Quando tutte le tabelle sono multi-year (hanno ``years``) e non c'è - hierarchy, il per-year ``run mart`` non ha nulla da elaborare. - """ - has_single_year = any(not t.years for t in cfg.mart.tables) - has_hierarchy = cfg.mart.hierarchy is not None - return has_single_year or has_hierarchy - - def _maybe_run_multi_year_mart( cfg, selected_years: list[int], @@ -446,7 +357,7 @@ def _maybe_run_multi_year_mart( ``sampling_active`` indica che il root e' stato spostato in ``{root}/smoke`` per via di ``--smoke``, ``--sample-rows`` o ``--sample-bytes``. """ - if not _has_multi_year_mart(cfg): + if not cfg.has_multi_year_mart: return if logger is None: logger = get_logger() @@ -803,13 +714,13 @@ def _run_batch( ) -> None: """Esegue piu' config in sequenza e stampa un report aggregato. - Legge un file di testo con un dataset.yml per riga (righe vuote e commenti - con # sono ignorati) e li esegue uno dopo l'altro. + Ogni config viene eseguito via ``_run_pipeline()`` — la stessa funzione + usata da ``toolkit run``. Il loop batch aggiunge solo la tabella riassuntiva. Args: batch_file: Path al file con lista di dataset.yml. - step: Step da eseguire (probe, raw, clean, mart, all). Default ``all``. - years: Non usato in batch (ogni config usa i propri anni configurati). + step: Step da eseguire (raw, clean, mart, all). Default ``all``. + years: Non usato (ogni config usa i propri anni). smoke: Alias per --sample-rows 1000 --sample-bytes 1048576. sample_rows: Limite righe in CLEAN. sample_bytes: Limite bytes in RAW. @@ -817,118 +728,84 @@ def _run_batch( json_output: Output JSON. dry_run: Solo plan senza esecuzione. """ - dry_flag = dry_run if isinstance(dry_run, bool) else False - sample_rows_final = 1000 if smoke else sample_rows - sample_bytes_final = 1048576 if smoke else sample_bytes - - configs_file = Path(batch_file) - config_paths = _read_config_list(configs_file) + config_paths = _read_config_list(Path(batch_file)) - rows: list[dict[str, str]] = [] - failures: list[dict[str, str]] = [] + rows: list[dict[str, Any]] = [] + failures: list[dict[str, Any]] = [] for config_path in config_paths: - config_started_at = perf_counter() + started_at = perf_counter() dataset_label = config_path.stem - try: - if smoke: - _cfg0, _logger0 = load_cfg_and_logger(str(config_path)) - if json_output: - silence_logger() - cfg, logger = load_cfg_and_logger( + # In modalita' --json, reindirizza stdout per evitare che i log + # del toolkit contaminino l'output JSON. + _stdout_ctx = ( + contextlib.redirect_stdout(io.StringIO()) if json_output else contextlib.nullcontext() + ) + + results: dict[str, Any] = {} + with _stdout_ctx: + try: + results = _run_pipeline( str(config_path), - root_override=str(_cfg0.root / "smoke"), + years=years, + step=step, + smoke=smoke, + sample_rows=sample_rows, + sample_bytes=sample_bytes, + root=root, + dry_run=dry_run, ) - else: - cfg, logger = load_cfg_and_logger(str(config_path)) - - if json_output: - silence_logger() - dataset_label = cfg.dataset - - for year in cfg.years: - run_started_at = perf_counter() + status = "SUCCESS" if results["status"] == "passed" else "FAILED" + dataset_label = results.get("dataset_name") or dataset_label + except Exception as exc: status = "FAILED" - try: - _run_ctx = silence_typer_echo() if json_output else contextlib.nullcontext() - with _run_ctx: - context = run_year( - cfg, - year, - step=step, - dry_run=dry_flag, - logger=logger, - sample_rows=sample_rows_final, - sample_bytes=sample_bytes_final, - ) - status = context.status - except Exception as exc: - failures.append( - { - "config": str(config_path), - "dataset": dataset_label, - "years": str(year), - "error": str(exc), - } - ) - finally: - rows.append( - build_row( - dataset=dataset_label, - config_path=str(config_path), - years=str(year), - step=step, - status=status, - duration=format_duration(perf_counter() - run_started_at), - ) - ) - except Exception as exc: - failures.append( + failures.append( + {"config": str(config_path), "dataset": dataset_label, "error": str(exc)} + ) + + rows.append( { - "config": str(config_path), "dataset": dataset_label, - "years": "-", - "error": str(exc), + "config": str(config_path), + "years": str(results.get("years")) if status == "SUCCESS" else "-", + "status": status, + "duration": f"{perf_counter() - started_at:.3f}s", } ) - rows.append( - build_row( - dataset=dataset_label, - config_path=str(config_path), - years="-", - step=step, - status="FAILED", - duration=format_duration(perf_counter() - config_started_at), - ) - ) if json_output: - report: dict[str, Any] = { - "summary": { - "total": len(rows), - "passed": sum(1 for r in rows if r["status"] in ("SUCCESS", "DRY_RUN")), - "failed": sum(1 for r in rows if r["status"] not in ("SUCCESS", "DRY_RUN")), - "duration_seconds": sum( - float(r["duration"].rstrip("s")) for r in rows if r["duration"] != "-" - ), - }, - "rows": rows, - "failures": failures, - } - typer.echo(json.dumps(report, indent=2, default=str)) + typer.echo( + json.dumps( + { + "summary": { + "total": len(rows), + "passed": sum(1 for r in rows if r["status"] == "SUCCESS"), + "failed": sum(1 for r in rows if r["status"] != "SUCCESS"), + }, + "rows": rows, + "failures": failures, + }, + indent=2, + default=str, + ) + ) else: - table_headers = ["dataset", "years", "step", "status", "duration"] - print_table(rows, table_headers) - + headers = ["dataset", "years", "status", "duration", "config"] + widths = {h: len(h) for h in headers} + for row in rows: + for h in headers: + widths[h] = max(widths[h], len(str(row.get(h, "")))) + typer.echo("Batch Report") + typer.echo(" ".join(h.ljust(widths[h]) for h in headers)) + typer.echo(" ".join("-" * widths[h] for h in headers)) + for row in rows: + typer.echo(" ".join(str(row.get(h, "")).ljust(widths[h]) for h in headers)) if failures: typer.echo("") typer.echo("Failures") - for failure in failures: - typer.echo( - f"- config={failure['config']} dataset={failure['dataset']} " - f"years={failure['years']} error={failure['error']}" - ) + for f in failures: + typer.echo(f"- config={f['config']} dataset={f['dataset']} error={f['error']}") if failures: raise typer.Exit(code=1) @@ -939,30 +816,34 @@ def _run_batch( # --------------------------------------------------------------------------- -def _execute_pipeline( +def _run_pipeline( config: str | None, - years: str | None, - smoke: bool, - sample_rows: int | None, - sample_bytes: int | None, - root: str | None, - json_output: bool, - dry_run: bool, -) -> None: + years: str | None = None, + step: str = "all", + smoke: bool = False, + sample_rows: int | None = None, + sample_bytes: int | None = None, + root: str | None = None, + dry_run: bool = False, +) -> dict[str, Any]: """Esegue pre-flight + support + raw → clean → mart. - Core della pipeline completa. Chiamata dal comando ``toolkit run`` - (default) e da ``run_full()`` (deprecato). + Versione pura senza output: non stampa nulla, non alza eccezioni CLI. + Restituisce un dict con esito, step e readiness. Args: config: Path/slug per dataset.yml, o None per auto-detect CWD. years: Anni separati da virgola, o None per tutti quelli configurati. + step: Step da eseguire (raw, clean, mart, all). Default ``all``. smoke: Se True, attiva --sample-rows 1000 --sample-bytes 1048576. sample_rows: Limite righe in CLEAN (LIMIT su output SQL). sample_bytes: Limite bytes in RAW (HTTP Range + troncamento). root: Override root output directory. - json_output: Se True, stampa report JSON su stdout. dry_run: Se True, solo plan senza esecuzione. + + Returns: + Dict con ``status`` (``passed``/``failed``), ``steps`` (per anno), + ``config``, ``years``, e readiness per ogni layer. """ dry_flag = dry_run if isinstance(dry_run, bool) else False @@ -970,7 +851,6 @@ def _execute_pipeline( sample_bytes_final = 1048576 if smoke else sample_bytes sample_mode = sample_rows_final is not None or sample_bytes_final is not None - # Qualsiasi forma di campionamento isola l'output in {root}/smoke sampling_active = sample_rows_final is not None or sample_bytes_final is not None root_override_final = root if sampling_active and not root and config is not None: @@ -983,6 +863,7 @@ def _execute_pipeline( results: dict[str, Any] = { "config": config, + "dataset_name": cfg.dataset, "years": selected_years, "steps": {}, "status": "passed", @@ -994,7 +875,6 @@ def _execute_pipeline( config_check = run_config_check(cfg, config) results["config_check"] = config_check if not config_check.get("ok", False): - # In dry-run il config check è meno severo: config senza fonti va bene if dry_flag: logger.warning( "Config: %s (dry-run, continua)", "; ".join(config_check.get("errors", [])) @@ -1002,9 +882,7 @@ def _execute_pipeline( else: logger.error("Config validation failed — aborting") results["status"] = "failed" - if json_output: - typer.echo(json.dumps(results, indent=2, default=str)) - raise typer.Exit(code=1) + return results for warn in config_check.get("warnings", []): logger.warning("Config: %s", warn) @@ -1023,7 +901,7 @@ def _execute_pipeline( sum(1 for s in preflight["sources"] if not s["reachable"]), ) - # Process support datasets (dichiarati in dataset.yml con support:) + # Process support datasets support_entries = cfg.support or [] if support_entries: logger.info( @@ -1034,7 +912,7 @@ def _execute_pipeline( logger.info("Support: %s — %s", entry.name, entry.config) if dry_flag: - typer.echo(f" [dry-run] support: {entry.name} — years={entry.years}") + logger.info(" [dry-run] support: %s — years=%s", entry.name, entry.years) continue try: @@ -1080,11 +958,11 @@ def _execute_pipeline( if results["status"] == "failed": break + # ── Candidate ──────────────────────────────────────────────────────── candidate_blocked = results["status"] == "failed" and not dry_flag - _candidate_exc: Exception | None = None if not candidate_blocked: - is_mart_only = _is_mart_only_cfg(cfg) - run_step = "mart" if is_mart_only else "all" + is_mart_only = cfg.is_mart_only + run_step = "mart" if (is_mart_only and step == "all") else step fail_on_error_flag = bool(cfg.validation.fail_on_error) for year in selected_years: @@ -1105,7 +983,6 @@ def _execute_pipeline( logger.error("Run %s year=%s fallito: %s", run_step, year, exc) results["steps"][str(year)] = {"run": "failed", "validate": "failed"} results["status"] = "failed" - _candidate_exc = exc break if not dry_flag: @@ -1124,9 +1001,7 @@ def _execute_pipeline( if not all_passed and fail_on_error_flag: results["status"] = "failed" - from toolkit.domain.readiness import ( - review_readiness as _review_readiness, - ) + from toolkit.domain.readiness import review_readiness as _review_readiness readiness = _review_readiness(config, year or None) results["steps"][str(year)]["readiness"] = readiness.get("readiness") @@ -1135,7 +1010,7 @@ def _execute_pipeline( results["steps"][str(year)]["checks_fail"] = readiness.get("fail_count", 0) results["steps"][str(year)]["layers"] = readiness.get("layers", {}) - if _candidate_exc is None and not dry_flag and _has_multi_year_mart(cfg): + if results["status"] == "passed" and not dry_flag and cfg.has_multi_year_mart: try: _maybe_run_multi_year_mart( cfg, @@ -1150,59 +1025,84 @@ def _execute_pipeline( if fail_on_error_flag: results["status"] = "failed" - if _candidate_exc is not None: - raise _candidate_exc + return results + + +def _execute_pipeline( + config: str | None, + years: str | None, + smoke: bool, + sample_rows: int | None, + sample_bytes: int | None, + root: str | None, + json_output: bool, + dry_run: bool, +) -> None: + """Wrapper CLI: esegue _run_pipeline e stampa output su stdout.""" + results = _run_pipeline( + config, + years, + step="all", + smoke=smoke, + sample_rows=sample_rows, + sample_bytes=sample_bytes, + root=root, + dry_run=dry_run, + ) if json_output: typer.echo(json.dumps(results, indent=2, default=str)) - else: - status = results["status"] - typer.echo(f"config: {config}") - typer.echo(f"years: {selected_years}") - typer.echo(f"status: {status}") - for y, s in results["steps"].items(): - typer.echo(f" {y}: run={s['run']} validate={s['validate']}") - lyrs = s.get("layers", {}) - for lname in ("raw", "clean", "mart"): - ln = lyrs.get(lname) or {} - lv = ln.get("validation") or {} - ok = lv.get("ok") - qs = lv.get("quality_score") - icon = "✅" if ok else ("🔴" if ok is False else "·") - parts = [] - if qs is not None: - parts.append(f"qs={qs}") - if lname == "raw": - pf = ln.get("profile") or {} - if pf.get("encoding"): - parts.append(f"encoding={pf['encoding']}") - if pf.get("delim"): - parts.append(f"delim={pf['delim']}") - pw = ln.get("profile_warnings") or [] - if pw: - parts.append(f"{len(pw)} warning") - elif lname == "clean": - rc = lv.get("row_count") or ln.get("row_count") - cc = lv.get("col_count") - if rc is not None: - parts.append(f"{rc} righe") - if cc is not None: - parts.append(f"{cc} colonne") - tr = ln.get("transition") or {} - if tr.get("row_drop_pct") is not None: - parts.append(f"raw->clean: {tr['row_drop_pct']}% righe") - elif lname == "mart": - tbl = ln.get("tables") or [] - ready = sum(1 for t in tbl if t.get("readable")) - parts.append(f"{ready}/{len(tbl)} tabelle") - typer.echo( - f" {lname}: {icon} {' '.join(parts)}" - if parts - else f" {lname}: {icon}" - ) + if results["status"] != "passed": + raise typer.Exit(code=1) + return + + status = results["status"] + typer.echo(f"config: {config}") + typer.echo(f"years: {results['years']}") + typer.echo(f"status: {status}") + for y, s in results.get("steps", {}).items(): + typer.echo(f" {y}: run={s['run']} validate={s['validate']}") + lyrs = s.get("layers", {}) + for lname in ("raw", "clean", "mart"): + ln = lyrs.get(lname) or {} + lv = ln.get("validation") or {} + ok = lv.get("ok") + qs = lv.get("quality_score") + icon = "✅" if ok else ("🔴" if ok is False else "·") + parts = [] + if qs is not None: + parts.append(f"qs={qs}") + if lname == "raw": + pf = ln.get("profile") or {} + if pf.get("encoding"): + parts.append(f"encoding={pf['encoding']}") + if pf.get("delim"): + parts.append(f"delim={pf['delim']}") + pw = ln.get("profile_warnings") or [] + if pw: + parts.append(f"{len(pw)} warning") + elif lname == "clean": + rc = lv.get("row_count") or ln.get("row_count") + cc = lv.get("col_count") + if rc is not None: + parts.append(f"{rc} righe") + if cc is not None: + parts.append(f"{cc} colonne") + tr = ln.get("transition") or {} + if tr.get("row_drop_pct") is not None: + parts.append(f"raw->clean: {tr['row_drop_pct']}% righe") + elif lname == "mart": + tbl = ln.get("tables") or [] + ready = sum(1 for t in tbl if t.get("readable")) + parts.append(f"{ready}/{len(tbl)} tabelle") typer.echo( - f" readiness: {s.get('readiness', '?')} ({s.get('checks_ok', 0)}/{s.get('checks', 0)})" + f" {lname}: {icon} {' '.join(parts)}" + if parts + else f" {lname}: {icon}" ) + typer.echo( + f" readiness: {s.get('readiness', '?')} ({s.get('checks_ok', 0)}/{s.get('checks', 0)})" + ) if results["status"] != "passed": raise typer.Exit(code=1) diff --git a/toolkit/cli/common.py b/toolkit/cli/common.py index 9d81d567..c89cf4ea 100644 --- a/toolkit/cli/common.py +++ b/toolkit/cli/common.py @@ -1,33 +1,13 @@ from __future__ import annotations -from typing import Any -from toolkit.core.config import load_config +from toolkit.core.config import ensure_dict, load_config from toolkit.core.logging import get_logger +# Re-export per backward compat dei consumer CLI +__all__ = ["dump_cfg_section", "load_cfg_and_logger"] -def dump_cfg_section(cfg_section: Any) -> Any: - """Convert config section to dict for functions expecting dict. - - Ordine: model_dump → to_dict (dataclass) → Mapping → altra iterabile → valore nudo. - Un dict non deve passare per il caso lista, altrimenti ``dump_cfg_section({"a": 1})`` - restituirebbe ``["a"]`` invece di ``{"a": 1}``. - """ - if hasattr(cfg_section, "model_dump"): - return cfg_section.model_dump( - mode="python", by_alias=True, exclude_none=True, exclude_unset=True - ) - if hasattr(cfg_section, "to_dict"): - return cfg_section.to_dict() - from dataclasses import asdict - - if hasattr(cfg_section, "__dataclass_fields__"): - return {k: v for k, v in asdict(cfg_section).items() if v is not None} - if isinstance(cfg_section, dict): - return cfg_section - if hasattr(cfg_section, "__iter__") and not isinstance(cfg_section, str): - return [dump_cfg_section(item) for item in cfg_section] - return cfg_section +dump_cfg_section = ensure_dict def load_cfg_and_logger( diff --git a/toolkit/cli/sql_dry_run.py b/toolkit/cli/sql_dry_run.py index 3ac68e2c..7f81d984 100644 --- a/toolkit/cli/sql_dry_run.py +++ b/toolkit/cli/sql_dry_run.py @@ -1,237 +1,13 @@ +"""SQL dry-run validation — DEPRECATO, importa da toolkit.core.sql_validation.""" + from __future__ import annotations -import logging -import re -from typing import Any +import warnings -import duckdb -from lab_connectors.duckdb import safe_connect +from toolkit.core.sql_validation import * # noqa: F403 -from toolkit.core.constants import RAW_INPUT_VIEW, CLEAN_INPUT_VIEW -from toolkit.clean.run import load_clean_sql -from toolkit.clean.sql_execute import _load_standard_macros -from toolkit.core.config import ensure_dict -from toolkit.core.paths import resolve_sql_path as _resolve_mart_sql_path -from toolkit.core.support import ( - check_support_path_drift, - flatten_support_template_ctx, - resolve_support_payloads, +warnings.warn( + "toolkit.cli.sql_dry_run è deprecato, importa da toolkit.core.sql_validation", + DeprecationWarning, + stacklevel=2, ) -from toolkit.core.template import build_runtime_template_ctx, render_template - -_logger = logging.getLogger("toolkit.cli.sql_dry_run") - -_QUOTED_IDENTIFIER_RE = re.compile(r'"([^"]+)"') -_BINDER_MISSING_COLUMN_RE = re.compile(r'Referenced column "([^"]+)" not found in FROM clause') - - -def _dedupe_preserve_order(items: list[str]) -> list[str]: - seen: set[str] = set() - result: list[str] = [] - for item in items: - if not item or item in seen: - continue - seen.add(item) - result.append(item) - return result - - -def _placeholder_columns(clean_cfg: dict[str, Any], sql: str) -> list[str]: - columns: list[str] = [] - read_cfg = clean_cfg.get("read") or {} - read_columns = read_cfg.get("columns") or {} - if isinstance(read_columns, dict): - columns.extend(str(name) for name in read_columns.keys()) - - # Fallback minimale: raccoglie identifier quoted dal SQL per costruire un - # raw_input placeholder abbastanza utile nel dry-run. E' deliberatamente - # approssimativo: puo' includere nomi non-colonna e non copre colonne non - # quotate se non sono gia' dichiarate in clean.read.columns. - columns.extend(match.group(1) for match in _QUOTED_IDENTIFIER_RE.finditer(sql)) - return _dedupe_preserve_order(columns) - - -def _quoted_identifier(name: str) -> str: - return '"' + name.replace('"', '""') + '"' - - -def _normalize_sql(sql: str) -> str: - return sql.strip().rstrip(";").strip() - - -def _create_placeholder_raw_input_with_columns( - con: duckdb.DuckDBPyConnection, - columns: list[str], -) -> None: - if columns: - projection = ", ".join(f"NULL::VARCHAR AS {_quoted_identifier(name)}" for name in columns) - else: - projection = "NULL::VARCHAR AS __dry_run_placeholder" - con.execute(f"CREATE OR REPLACE VIEW {RAW_INPUT_VIEW} AS SELECT {projection} LIMIT 0") - - -def _extract_missing_binder_column(exc: Exception) -> str | None: - match = _BINDER_MISSING_COLUMN_RE.search(str(exc)) - if not match: - return None - return match.group(1) - - -def _build_clean_preview( - cfg, - *, - year: int, - con: duckdb.DuckDBPyConnection, - support_cfg: list[dict[str, Any]] | None = None, - dry_run: bool = False, -) -> None: - clean_cfg_ = ensure_dict(cfg.clean) - clean_sql_path, clean_sql, _ = load_clean_sql( - clean_cfg_, - dataset=cfg.dataset, - year=year, - root=cfg.root, - base_dir=cfg.base_dir, - support_cfg=support_cfg, - ) - - clean_sql = _normalize_sql(clean_sql) - columns = _placeholder_columns(clean_cfg_, clean_sql) - - # Pre-calcola i path attesi del support per gestire IOError in dry-run - support_paths: list[str] = [] - support_payloads_drift: list[dict[str, Any]] = [] - if dry_run and support_cfg: - try: - sp_payloads = resolve_support_payloads(support_cfg, require_exists=False, smoke=False) - support_paths = _all_support_expected_paths(sp_payloads) - support_payloads_drift = sp_payloads - except Exception: - pass - - # Anti-path-drift: verifica che il SQL non referenzi support dataset - # via path hardcoded invece di {support.NAME.mart} - if support_payloads_drift: - raw_sql_text = clean_sql_path.read_text(encoding="utf-8") - drift_warnings = check_support_path_drift( - raw_sql_text, support_payloads_drift, sql_label=str(clean_sql_path) - ) - for w in drift_warnings: - _logger.warning("PATH DRIFT: %s", w) - - # Fallback incrementale: se il clean.sql usa colonne raw non quotate e non - # dichiarate in clean.read.columns, il binder di DuckDB ci dice il nome - # mancante. Lo aggiungiamo al placeholder e riproviamo, cosi' il dry-run - # evita falsi positivi banali senza provare a parsare SQL completo. - for _ in range(25): - _create_placeholder_raw_input_with_columns(con, columns) - try: - con.execute( - f"CREATE OR REPLACE TABLE __dry_run_clean_preview AS SELECT * FROM ({clean_sql}) AS q LIMIT 0" - ) - return - except Exception as exc: - err_msg = str(exc) - # In dry-run, read_parquet su file support non ancora generato è OK - if dry_run and "No files found that match the pattern" in err_msg: - if support_paths and any(sp in err_msg for sp in support_paths): - # Crea un placeholder minimo per non bloccare mart validation - con.execute( - "CREATE OR REPLACE TABLE __dry_run_clean_preview AS SELECT NULL::VARCHAR AS __support_placeholder LIMIT 0" - ) - return - - missing = _extract_missing_binder_column(exc) - if missing and missing not in columns: - columns.append(missing) - continue - raise ValueError(f"CLEAN SQL dry-run failed ({clean_sql_path}): {exc}") from exc - - raise ValueError( - f"CLEAN SQL dry-run failed ({clean_sql_path}): exceeded placeholder inference attempts" - ) - - -def _all_support_expected_paths(support_payloads: list[dict[str, Any]]) -> list[str]: - """All support mart output paths attesi (anche se non esistono ancora).""" - paths: list[str] = [] - for entry in support_payloads: - paths.extend(entry.get("outputs", [])) - return paths - - -def _validate_mart_sql( - cfg, *, year: int, con: duckdb.DuckDBPyConnection, dry_run: bool = False -) -> None: - clean_cfg_ = ensure_dict(cfg.clean) - mart_cfg_ = ensure_dict(cfg.mart) - if clean_cfg_.get("sql"): - con.execute( - f"CREATE OR REPLACE VIEW {CLEAN_INPUT_VIEW} AS SELECT * FROM __dry_run_clean_preview" - ) - - tables = mart_cfg_.get("tables") or [] - # In dry-run: require_exists=False per non bloccare candidate senza support - # ancora generati. Se DuckDB fallisce su read_parquet (file non esiste), - # il dry-run lo registra come avviso ma non blocca. - support_payloads = resolve_support_payloads( - ensure_dict(cfg.support), require_exists=not dry_run - ) - template_ctx = build_runtime_template_ctx( - dataset=cfg.dataset, - year=year, - root=cfg.root, - base_dir=cfg.base_dir, - support=flatten_support_template_ctx(support_payloads), - ) - - for table in tables: - name = table.get("name") - sql_ref = table.get("sql") - sql_path = _resolve_mart_sql_path(sql_ref, base_dir=cfg.base_dir) - raw_sql_text = sql_path.read_text(encoding="utf-8") - # Anti-path-drift: verifica che il SQL non referenzi support dataset - # via path hardcoded invece di {support.NAME.mart} - if support_payloads: - drift_warnings = check_support_path_drift( - raw_sql_text, support_payloads, sql_label=f"{sql_path} ({name})" - ) - for w in drift_warnings: - _logger.warning("PATH DRIFT: %s", w) - sql = _normalize_sql(render_template(raw_sql_text, template_ctx)) - try: - con.execute(f"EXPLAIN SELECT * FROM ({sql}) AS q LIMIT 0") - except Exception as exc: - err_msg = str(exc) - # In dry-run, DuckDB puo' fallire su read_parquet se il file support - # non esiste ancora (placeholder {support.*.mart}). Verifichiamo che - # l'errore riguardi un path support atteso, non un qualsiasi IO Error. - if dry_run and "No files found that match the pattern" in err_msg: - support_paths = _all_support_expected_paths(support_payloads) - if support_paths and any(sp in err_msg for sp in support_paths): - continue - raise ValueError(f"MART SQL dry-run failed ({name}, {sql_path}): {exc}") from exc - - -def validate_sql_dry_run(cfg, *, year: int, layers: list[str], dry_run: bool = False) -> None: - if not any(layer in {"clean", "mart"} for layer in layers): - return - - # ── Config check (zero network I/O, sempre) ───────────────────────── - from toolkit.domain.preflight import run_config_check - - config_check = run_config_check(cfg, cfg.base_dir / "dataset.yml") - if not config_check.get("ok", False): - for err in config_check.get("errors", []): - _logger.error("CONFIG ERR: %s", err) - for warn in config_check.get("warnings", []): - _logger.warning("CONFIG WARN: %s", warn) - - with safe_connect() as con: - _load_standard_macros(con, logger=None) - if cfg.clean.sql: - _build_clean_preview( - cfg, year=year, con=con, support_cfg=ensure_dict(cfg.support), dry_run=dry_run - ) - if "mart" in layers: - _validate_mart_sql(cfg, year=year, con=con, dry_run=dry_run) diff --git a/toolkit/core/config.py b/toolkit/core/config.py index 616b5d6e..9f4b72b0 100644 --- a/toolkit/core/config.py +++ b/toolkit/core/config.py @@ -542,6 +542,23 @@ def resolve(self, rel_path: str | Path) -> Path: return p return (self.base_dir / p).resolve() + @property + def is_mart_only(self) -> bool: + """``True`` se il dataset ha solo configurazione MART (no CLEAN SQL).""" + return not bool(self.clean.sql) + + @property + def has_multi_year_mart(self) -> bool: + """``True`` se una tabella MART ha years espliciti (multi-year).""" + return any(t.years for t in self.mart.tables) + + @property + def has_single_year_mart(self) -> bool: + """``True`` se una tabella MART non ha years (per-year) o c'è hierarchy.""" + has_single_year = any(not t.years for t in self.mart.tables) + has_hierarchy = self.mart.hierarchy is not None + return has_single_year or has_hierarchy + # Backward compat aliases ToolkitConfig = PipelineConfig diff --git a/toolkit/core/probe.py b/toolkit/core/probe.py index b79bc75c..33e9150d 100644 --- a/toolkit/core/probe.py +++ b/toolkit/core/probe.py @@ -217,3 +217,30 @@ def __enter__(self) -> ProbePool: def __exit__(self, *args: Any) -> None: self.close() + + +# --------------------------------------------------------------------------- +# Probe format hint +# --------------------------------------------------------------------------- + +_PROBE_FORMATS: dict[str, str] = { + "text/csv": "CSV", + "text/tab-separated-values": "TSV", + "application/json": "JSON", + "application/xml": "XML", + "application/zip": "ZIP", + "application/gzip": "GZ", + "application/pdf": "PDF", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": "XLSX", + "application/vnd.ms-excel": "XLS", + "application/vnd.oasis.opendocument.spreadsheet": "ODS", + "text/html": "HTML", +} + + +def probe_fmt(content_type: str | None) -> str: + """Riduce un content-type MIME a un formato leggibile (es. XLSX, CSV).""" + if not content_type: + return "?" + base = content_type.split(";")[0].strip().lower() + return _PROBE_FORMATS.get(base, base) diff --git a/toolkit/core/sql_validation.py b/toolkit/core/sql_validation.py new file mode 100644 index 00000000..4f8560c3 --- /dev/null +++ b/toolkit/core/sql_validation.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import logging +import re +from typing import Any + +import duckdb +from lab_connectors.duckdb import safe_connect + +from toolkit.core.constants import RAW_INPUT_VIEW, CLEAN_INPUT_VIEW +from toolkit.clean.run import load_clean_sql +from toolkit.clean.sql_execute import _load_standard_macros +from toolkit.core.config import ensure_dict +from toolkit.core.paths import resolve_sql_path as _resolve_mart_sql_path +from toolkit.core.sql_utils import q_ident +from toolkit.core.support import ( + check_support_path_drift, + flatten_support_template_ctx, + resolve_support_payloads, +) +from toolkit.core.template import build_runtime_template_ctx, render_template + +_logger = logging.getLogger("toolkit.core.sql_validation") + +_QUOTED_IDENTIFIER_RE = re.compile(r'"([^"]+)"') +_BINDER_MISSING_COLUMN_RE = re.compile(r'Referenced column "([^"]+)" not found in FROM clause') + + +def _dedupe_preserve_order(items: list[str]) -> list[str]: + seen: set[str] = set() + result: list[str] = [] + for item in items: + if not item or item in seen: + continue + seen.add(item) + result.append(item) + return result + + +def _placeholder_columns(clean_cfg: dict[str, Any], sql: str) -> list[str]: + columns: list[str] = [] + read_cfg = clean_cfg.get("read") or {} + read_columns = read_cfg.get("columns") or {} + if isinstance(read_columns, dict): + columns.extend(str(name) for name in read_columns.keys()) + + # Fallback minimale: raccoglie identifier quoted dal SQL per costruire un + # raw_input placeholder abbastanza utile nel dry-run. E' deliberatamente + # approssimativo: puo' includere nomi non-colonna e non copre colonne non + # quotate se non sono gia' dichiarate in clean.read.columns. + columns.extend(match.group(1) for match in _QUOTED_IDENTIFIER_RE.finditer(sql)) + return _dedupe_preserve_order(columns) + + +def _normalize_sql(sql: str) -> str: + return sql.strip().rstrip(";").strip() + + +def _create_placeholder_raw_input_with_columns( + con: duckdb.DuckDBPyConnection, + columns: list[str], +) -> None: + if columns: + projection = ", ".join(f"NULL::VARCHAR AS {q_ident(name)}" for name in columns) + else: + projection = "NULL::VARCHAR AS __dry_run_placeholder" + con.execute(f"CREATE OR REPLACE VIEW {RAW_INPUT_VIEW} AS SELECT {projection} LIMIT 0") + + +def _extract_missing_binder_column(exc: Exception) -> str | None: + match = _BINDER_MISSING_COLUMN_RE.search(str(exc)) + if not match: + return None + return match.group(1) + + +def _build_clean_preview( + cfg, + *, + year: int, + con: duckdb.DuckDBPyConnection, + support_cfg: list[dict[str, Any]] | None = None, + dry_run: bool = False, +) -> None: + clean_cfg_ = ensure_dict(cfg.clean) + clean_sql_path, clean_sql, _ = load_clean_sql( + clean_cfg_, + dataset=cfg.dataset, + year=year, + root=cfg.root, + base_dir=cfg.base_dir, + support_cfg=support_cfg, + ) + + clean_sql = _normalize_sql(clean_sql) + columns = _placeholder_columns(clean_cfg_, clean_sql) + + # Pre-calcola i path attesi del support per gestire IOError in dry-run + support_paths: list[str] = [] + support_payloads_drift: list[dict[str, Any]] = [] + if dry_run and support_cfg: + try: + sp_payloads = resolve_support_payloads(support_cfg, require_exists=False, smoke=False) + support_paths = _all_support_expected_paths(sp_payloads) + support_payloads_drift = sp_payloads + except Exception: + pass + + # Anti-path-drift: verifica che il SQL non referenzi support dataset + # via path hardcoded invece di {support.NAME.mart} + if support_payloads_drift: + raw_sql_text = clean_sql_path.read_text(encoding="utf-8") + drift_warnings = check_support_path_drift( + raw_sql_text, support_payloads_drift, sql_label=str(clean_sql_path) + ) + for w in drift_warnings: + _logger.warning("PATH DRIFT: %s", w) + + # Fallback incrementale: se il clean.sql usa colonne raw non quotate e non + # dichiarate in clean.read.columns, il binder di DuckDB ci dice il nome + # mancante. Lo aggiungiamo al placeholder e riproviamo, cosi' il dry-run + # evita falsi positivi banali senza provare a parsare SQL completo. + for _ in range(25): + _create_placeholder_raw_input_with_columns(con, columns) + try: + con.execute( + f"CREATE OR REPLACE TABLE __dry_run_clean_preview AS SELECT * FROM ({clean_sql}) AS q LIMIT 0" + ) + return + except Exception as exc: + err_msg = str(exc) + # In dry-run, read_parquet su file support non ancora generato è OK + if dry_run and "No files found that match the pattern" in err_msg: + if support_paths and any(sp in err_msg for sp in support_paths): + # Crea un placeholder minimo per non bloccare mart validation + con.execute( + "CREATE OR REPLACE TABLE __dry_run_clean_preview AS SELECT NULL::VARCHAR AS __support_placeholder LIMIT 0" + ) + return + + missing = _extract_missing_binder_column(exc) + if missing and missing not in columns: + columns.append(missing) + continue + raise ValueError(f"CLEAN SQL dry-run failed ({clean_sql_path}): {exc}") from exc + + raise ValueError( + f"CLEAN SQL dry-run failed ({clean_sql_path}): exceeded placeholder inference attempts" + ) + + +def _all_support_expected_paths(support_payloads: list[dict[str, Any]]) -> list[str]: + """All support mart output paths attesi (anche se non esistono ancora).""" + paths: list[str] = [] + for entry in support_payloads: + paths.extend(entry.get("outputs", [])) + return paths + + +def _validate_mart_sql( + cfg, *, year: int, con: duckdb.DuckDBPyConnection, dry_run: bool = False +) -> None: + clean_cfg_ = ensure_dict(cfg.clean) + mart_cfg_ = ensure_dict(cfg.mart) + if clean_cfg_.get("sql"): + con.execute( + f"CREATE OR REPLACE VIEW {CLEAN_INPUT_VIEW} AS SELECT * FROM __dry_run_clean_preview" + ) + + tables = mart_cfg_.get("tables") or [] + # In dry-run: require_exists=False per non bloccare candidate senza support + # ancora generati. Se DuckDB fallisce su read_parquet (file non esiste), + # il dry-run lo registra come avviso ma non blocca. + support_payloads = resolve_support_payloads( + ensure_dict(cfg.support), require_exists=not dry_run + ) + template_ctx = build_runtime_template_ctx( + dataset=cfg.dataset, + year=year, + root=cfg.root, + base_dir=cfg.base_dir, + support=flatten_support_template_ctx(support_payloads), + ) + + for table in tables: + name = table.get("name") + sql_ref = table.get("sql") + sql_path = _resolve_mart_sql_path(sql_ref, base_dir=cfg.base_dir) + raw_sql_text = sql_path.read_text(encoding="utf-8") + # Anti-path-drift: verifica che il SQL non referenzi support dataset + # via path hardcoded invece di {support.NAME.mart} + if support_payloads: + drift_warnings = check_support_path_drift( + raw_sql_text, support_payloads, sql_label=f"{sql_path} ({name})" + ) + for w in drift_warnings: + _logger.warning("PATH DRIFT: %s", w) + sql = _normalize_sql(render_template(raw_sql_text, template_ctx)) + try: + con.execute(f"EXPLAIN SELECT * FROM ({sql}) AS q LIMIT 0") + except Exception as exc: + err_msg = str(exc) + # In dry-run, DuckDB puo' fallire su read_parquet se il file support + # non esiste ancora (placeholder {support.*.mart}). Verifichiamo che + # l'errore riguardi un path support atteso, non un qualsiasi IO Error. + if dry_run and "No files found that match the pattern" in err_msg: + support_paths = _all_support_expected_paths(support_payloads) + if support_paths and any(sp in err_msg for sp in support_paths): + continue + raise ValueError(f"MART SQL dry-run failed ({name}, {sql_path}): {exc}") from exc + + +def validate_sql_dry_run(cfg, *, year: int, layers: list[str], dry_run: bool = False) -> None: + if not any(layer in {"clean", "mart"} for layer in layers): + return + + # ── Config check (zero network I/O, sempre) ───────────────────────── + from toolkit.domain.preflight import run_config_check + + config_check = run_config_check(cfg, cfg.base_dir / "dataset.yml") + if not config_check.get("ok", False): + for err in config_check.get("errors", []): + _logger.error("CONFIG ERR: %s", err) + for warn in config_check.get("warnings", []): + _logger.warning("CONFIG WARN: %s", warn) + + with safe_connect() as con: + _load_standard_macros(con, logger=None) + if cfg.clean.sql: + _build_clean_preview( + cfg, year=year, con=con, support_cfg=ensure_dict(cfg.support), dry_run=dry_run + ) + if "mart" in layers: + _validate_mart_sql(cfg, year=year, con=con, dry_run=dry_run) From 00bce14909fc6133d903006b3698a353c849421f Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Fri, 31 Jul 2026 09:30:28 +0100 Subject: [PATCH 2/2] fix: exit code batch su run falliti + ripristino DRY_RUN report Review PR #441: - Critical: _run_batch usciva con 0 quando _run_pipeline ritornava status=failed senza eccezione (config check fallito). Ora exit 1 se failures o qualsiasi riga non SUCCESS/DRY_RUN. - Medium: report batch ri-segnala DRY_RUN (era perso dopo refactor). summary.passed conta SUCCESS+DRY_RUN. - Medium: test dry-run ripristinati con assert sui messaggi errore (normalizzati per il logger rich che spezza le righe). - Low: ensure_dict documentato (filtro None == vecchio dump_cfg_section). - Test batch: config test con read.columns esplicite per SQL validation dry-run. --- tests/test_batch_cli.py | 35 +++++++++++++++++++++++++++++++++-- tests/test_run_dry_run.py | 16 ++++++++++++++++ toolkit/cli/cmd_run.py | 11 +++++++---- toolkit/core/config.py | 5 +++++ 4 files changed, 61 insertions(+), 6 deletions(-) diff --git a/tests/test_batch_cli.py b/tests/test_batch_cli.py index 11e097c4..0f0cb1cc 100644 --- a/tests/test_batch_cli.py +++ b/tests/test_batch_cli.py @@ -71,6 +71,10 @@ def _write_batch_project(project_dir: Path, dataset: str, year: int) -> Path: decimal: "," mode: explicit include: {dataset}_{year}.csv + columns: + comune: VARCHAR + anno: INTEGER + valore: DOUBLE required_columns: comune validate: not_null: valore @@ -266,6 +270,7 @@ def test_batch_dry_run_with_json(tmp_path: Path) -> None: assert result.exit_code == 0 report = json.loads(result.stdout) assert report["summary"]["total"] == 1 + assert report["summary"]["passed"] == 1 assert report["rows"][0]["dataset"] == "batch_dry_json" # Nessun file creato (dry-run) @@ -275,7 +280,7 @@ def test_batch_dry_run_with_json(tmp_path: Path) -> None: @pytest.mark.policy def test_batch_step_probe_dry_run_reports_dry_run(tmp_path: Path) -> None: - """``toolkit run --batch --dry-run --json`` non crea file.""" + """``toolkit run --batch --dry-run --json`` riporta DRY_RUN e non crea file.""" project = tmp_path / "project" _write_batch_project(project, "batch_probe_dry", 2023) configs_file = _write_configs_file(tmp_path, "project") @@ -290,7 +295,8 @@ def test_batch_step_probe_dry_run_reports_dry_run(tmp_path: Path) -> None: assert result.exit_code == 0 report = json.loads(result.stdout) assert report["summary"]["total"] == 1 - assert report["rows"][0]["status"] in ("SUCCESS", "FAILED") + assert report["summary"]["passed"] == 1 + assert report["rows"][0]["status"] == "DRY_RUN" # Dry-run non crea output fisici raw_out = project / "out" / "data" / "raw" / "batch_probe_dry" / "2023" @@ -371,3 +377,28 @@ def test_run_batch_end_to_end(tmp_path: Path) -> None: # Verifica output fisici assert (project_a / "out" / "data" / "mart" / "e2e_a" / "2022" / "mart_totali.parquet").exists() assert (project_b / "out" / "data" / "mart" / "e2e_b" / "2023" / "mart_totali.parquet").exists() + + +@pytest.mark.contract +def test_run_batch_exit_nonzero_when_config_invalid(tmp_path: Path) -> None: + """``toolkit run --batch`` con config non bootstrappata deve uscire con codice != 0. + + Regressione: _run_pipeline ritorna ``status="failed"`` senza eccezione + quando il run fallisce (es. clean.sql mancante). _run_batch deve + propagare l'esito fallito (riga FAILED) come exit code 1, non uscire con 0. + """ + project = tmp_path / "project" + _write_batch_project(project, "batch_invalid", 2023) + # Rimuove clean.sql: il run fallisce con "CLEAN SQL file not found" + (project / "sql" / "clean.sql").unlink() + configs_file = _write_configs_file(tmp_path, "project") + + runner = CliRunner() + result = runner.invoke( + app, + ["run", "--batch", str(configs_file)], + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "FAILED" in result.output diff --git a/tests/test_run_dry_run.py b/tests/test_run_dry_run.py index 47906fa7..fad237ab 100644 --- a/tests/test_run_dry_run.py +++ b/tests/test_run_dry_run.py @@ -4,6 +4,7 @@ import json import logging +import re import time from pathlib import Path @@ -16,6 +17,11 @@ from tests.helpers import make_dataset_yml, make_standard_sql +def _normalized(text: str) -> str: + """Collassa spazi e newline (il logger rich spezza le righe).""" + return re.sub(r"\s+", " ", text) + + # ── Basic patterns ────────────────────────────────────────────────────────── @@ -66,6 +72,12 @@ def test_run_dry_run_fails_on_clean_sql_syntax_error(tmp_path: Path, runner) -> result = runner.invoke(app, ["run", "--config", str(config_path), "--dry-run"]) assert result.exit_code != 0 + # Il logger rich spezza le righe e inserisce il path del file: + # "CLEAN SQL cmd_run.py:986 dry-run failed (...)". Verifichiamo le parti. + normalized = _normalized(result.output) + assert "CLEAN SQL" in normalized + assert "dry-run failed" in normalized + assert "Parser Error" in normalized @pytest.mark.policy @@ -89,6 +101,10 @@ def test_run_dry_run_fails_on_mart_sql_binding_error(tmp_path: Path, runner) -> result = runner.invoke(app, ["run", "--config", str(config_path), "--dry-run"]) assert result.exit_code != 0 + normalized = _normalized(result.output) + assert "MART SQL" in normalized + assert "dry-run failed" in normalized + assert "Binder Error" in normalized @pytest.mark.policy diff --git a/toolkit/cli/cmd_run.py b/toolkit/cli/cmd_run.py index 1134b61b..f22dbc68 100644 --- a/toolkit/cli/cmd_run.py +++ b/toolkit/cli/cmd_run.py @@ -757,6 +757,9 @@ def _run_batch( dry_run=dry_run, ) status = "SUCCESS" if results["status"] == "passed" else "FAILED" + # In dry-run, un esito ok e' un piano valido, non una run eseguita + if status == "SUCCESS" and dry_run: + status = "DRY_RUN" dataset_label = results.get("dataset_name") or dataset_label except Exception as exc: status = "FAILED" @@ -768,7 +771,7 @@ def _run_batch( { "dataset": dataset_label, "config": str(config_path), - "years": str(results.get("years")) if status == "SUCCESS" else "-", + "years": str(results.get("years")) if status in ("SUCCESS", "DRY_RUN") else "-", "status": status, "duration": f"{perf_counter() - started_at:.3f}s", } @@ -780,8 +783,8 @@ def _run_batch( { "summary": { "total": len(rows), - "passed": sum(1 for r in rows if r["status"] == "SUCCESS"), - "failed": sum(1 for r in rows if r["status"] != "SUCCESS"), + "passed": sum(1 for r in rows if r["status"] in ("SUCCESS", "DRY_RUN")), + "failed": sum(1 for r in rows if r["status"] not in ("SUCCESS", "DRY_RUN")), }, "rows": rows, "failures": failures, @@ -807,7 +810,7 @@ def _run_batch( for f in failures: typer.echo(f"- config={f['config']} dataset={f['dataset']} error={f['error']}") - if failures: + if failures or any(r["status"] not in ("SUCCESS", "DRY_RUN") for r in rows): raise typer.Exit(code=1) diff --git a/toolkit/core/config.py b/toolkit/core/config.py index 9f4b72b0..97c32e93 100644 --- a/toolkit/core/config.py +++ b/toolkit/core/config.py @@ -574,6 +574,11 @@ def ensure_dict(cfg: Any) -> Any: """Convert a config section to a plain dict for runner layers. Handles dataclasses, old Pydantic models, dicts, and lists. + + Nota: per le dataclass i campi a ``None`` vengono esclusi + (``{k: v ... if v is not None}``). I consumer devono usare + ``.get()``, non ``in``/``.keys()``. Questo replica il comportamento + del vecchio ``cli.common.dump_cfg_section`` (rimosso). """ if hasattr(cfg, "to_dict"): return cfg.to_dict()