From 3d200628f1ff44b0b51381fad69c788e1cc69ec0 Mon Sep 17 00:00:00 2001 From: Neukz Date: Sun, 21 Jun 2026 13:32:36 +0200 Subject: [PATCH] feature: add DAG failure handling and ingestion summary --- dags/repolytics_daily.py | 59 +++++++++++++++++++++++----- src/repolytics/ingestion/pipeline.py | 43 ++++++++++++++------ 2 files changed, 80 insertions(+), 22 deletions(-) diff --git a/dags/repolytics_daily.py b/dags/repolytics_daily.py index 13b7ea2..d7ea0cb 100644 --- a/dags/repolytics_daily.py +++ b/dags/repolytics_daily.py @@ -6,6 +6,7 @@ serializes everything so no two tasks open the single-writer DuckDB file at once. """ +import logging import os from datetime import timedelta from pathlib import Path @@ -19,9 +20,25 @@ RenderConfig, ) +logger = logging.getLogger(__name__) + # Location of the dbt project inside the container DBT_PROJECT_DIR = Path(os.environ.get("DBT_PROJECT_DIR", "/opt/airflow/dbt")) + +def log_task_failure(context) -> None: + """Structured failure log for any task (wired via ``default_args``).""" + ti = context.get("task_instance") + logger.error( + "Task failed: dag=%s task=%s run_id=%s try=%s exception=%r", + getattr(ti, "dag_id", None), + getattr(ti, "task_id", None), + context.get("run_id"), + getattr(ti, "try_number", None), + context.get("exception"), + ) + + profile_config = ProfileConfig( profile_name="repolytics", target_name="dev", @@ -34,23 +51,33 @@ schedule="@daily", catchup=False, max_active_tasks=1, # DuckDB is single-writer: never run two dbt tasks at once. - default_args={"retries": 2, "retry_delay": timedelta(minutes=5)}, + default_args={ + "retries": 2, + "retry_delay": timedelta(minutes=5), + "on_failure_callback": log_task_failure, + }, tags=["repolytics", "elt"], ) def repolytics_daily(): - @task - def ingest_github() -> None: - """Run GitHub ingestion into the DuckDB `raw` dataset.""" + @task(multiple_outputs=False) + def ingest_github() -> dict[str, int]: + """Run GitHub ingestion into the DuckDB `raw` dataset. + + Returns per-table row counts (pushed to XCom for the summary task). + """ from repolytics.ingestion.pipeline import run_github - run_github() + return run_github() - @task - def ingest_pypi() -> None: - """Run PyPI ingestion into the DuckDB `raw` dataset.""" + @task(multiple_outputs=False) + def ingest_pypi() -> dict[str, int]: + """Run PyPI ingestion into the DuckDB `raw` dataset. + + Returns per-table row counts (pushed to XCom for the summary task). + """ from repolytics.ingestion.pipeline import run_pypi - run_pypi() + return run_pypi() transform = DbtTaskGroup( group_id="transform", @@ -68,7 +95,19 @@ def ingest_pypi() -> None: ), ) - [ingest_github(), ingest_pypi()] >> transform + @task + def summary(github_counts: dict[str, int], pypi_counts: dict[str, int]) -> None: + """Log per-source row counts for the run.""" + logger.info( + "Ingestion summary - GitHub: %s | PyPI: %s", + github_counts, + pypi_counts, + ) + + github_counts = ingest_github() + pypi_counts = ingest_pypi() + + [github_counts, pypi_counts] >> transform >> summary(github_counts, pypi_counts) repolytics_daily() diff --git a/src/repolytics/ingestion/pipeline.py b/src/repolytics/ingestion/pipeline.py index d11ae1b..88d0537 100644 --- a/src/repolytics/ingestion/pipeline.py +++ b/src/repolytics/ingestion/pipeline.py @@ -1,4 +1,6 @@ -"""dlt pipeline wiring GitHub + PyPI ingestion into the DuckDB `raw` dataset.""" +"""dlt pipelines wiring GitHub and PyPI ingestion into the DuckDB `raw` dataset.""" + +import logging import dlt @@ -6,6 +8,8 @@ from repolytics.ingestion.github_source import github_source from repolytics.ingestion.pypi_source import pypi_source +logger = logging.getLogger(__name__) + def build_pipeline(settings: Settings) -> dlt.Pipeline: """Create the DuckDB-backed dlt pipeline targeting the `raw` dataset.""" @@ -16,44 +20,59 @@ def build_pipeline(settings: Settings) -> dlt.Pipeline: ) -def run_github(settings: Settings | None = None) -> None: +def _table_row_counts(pipeline: dlt.Pipeline) -> dict[str, int]: + """Rows loaded per table in the last run, excluding dlt's internal tables. + + Reads `last_trace.last_normalize_info.row_counts` (table name -> count) and + drops the `_dlt_*` bookkeeping tables so the summary reflects only ingested data. + """ + counts = pipeline.last_trace.last_normalize_info.row_counts + return {table: n for table, n in counts.items() if not table.startswith("_dlt")} + + +def run_github(settings: Settings | None = None) -> dict[str, int]: """Run GitHub ingestion into the configured DuckDB warehouse. - No-ops (logs and returns) when no repos are configured, so an empty repo list - succeeds as a skip rather than failing. + No-ops (logs and returns `{}`) when no repos are configured, so an empty repo + list succeeds as a skip rather than failing. Returns the per-table row counts + loaded in this run. """ settings = settings or get_settings() repos = settings.target_repos if not repos: - print(f"No repos to ingest - check {settings.projects_file}") - return + logger.warning("No repos to ingest - check %s", settings.projects_file) + return {} settings.duckdb_path.parent.mkdir(parents=True, exist_ok=True) pipeline = build_pipeline(settings) source = github_source(repos, settings.github_token.get_secret_value()) - print(pipeline.run(source)) + logger.info("GitHub ingestion complete: %s", pipeline.run(source)) + return _table_row_counts(pipeline) -def run_pypi(settings: Settings | None = None) -> None: +def run_pypi(settings: Settings | None = None) -> dict[str, int]: """Run PyPI ingestion into the configured DuckDB warehouse. - No-ops (logs and returns) when no packages are configured. + No-ops (logs and returns `{}`) when no packages are configured. Returns the + per-table row counts loaded in this run. """ settings = settings or get_settings() packages = settings.packages if not packages: - print(f"No packages to ingest - check {settings.projects_file}") - return + logger.warning("No packages to ingest - check %s", settings.projects_file) + return {} settings.duckdb_path.parent.mkdir(parents=True, exist_ok=True) pipeline = build_pipeline(settings) - print(pipeline.run(pypi_source(packages))) + logger.info("PyPI ingestion complete: %s", pipeline.run(pypi_source(packages))) + return _table_row_counts(pipeline) if __name__ == "__main__": # Run GitHub + PyPI ingestion (CLI convenience wrapper for both sources). + logging.basicConfig(level=logging.INFO) settings = get_settings() run_github(settings) run_pypi(settings)