feat: add DAG failure handling and ingestion summary

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()

src/repolytics/ingestion/pipeline.py

@@ -1,11 +1,15 @@
-"""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
 
 from repolytics.config import Settings, get_settings
 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)