From 6e54dd7530fc6bf2aaf7c41706156d453f10d6a6 Mon Sep 17 00:00:00 2001
From: Neukz <kacperneumann1@gmail.com>
Date: Sat, 20 Jun 2026 22:24:11 +0200
Subject: [PATCH] refactor: split ingestion into per-source github/pypi tasks

---
 Taskfile.yml                         | 14 +++++++--
 dags/repolytics_daily.py             | 24 ++++++++++-----
 src/repolytics/ingestion/pipeline.py | 44 +++++++++++++++++++---------
 3 files changed, 58 insertions(+), 24 deletions(-)

diff --git a/Taskfile.yml b/Taskfile.yml
index 3172d4b..32af86b 100644
--- a/Taskfile.yml
+++ b/Taskfile.yml
@@ -65,6 +65,16 @@ tasks:
     cmds:
       - uv run python -m repolytics.ingestion.pipeline
 
+  dlt:ingest:github:
+    desc: Run only GitHub ingestion into the DuckDB raw schema
+    cmds:
+      - uv run python -c "from repolytics.ingestion.pipeline import run_github; run_github()"
+
+  dlt:ingest:pypi:
+    desc: Run only PyPI ingestion into the DuckDB raw schema
+    cmds:
+      - uv run python -c "from repolytics.ingestion.pipeline import run_pypi; run_pypi()"
+
   # ----- dbt -----
   dbt:seed:
     desc: Load CSV seeds into the warehouse
@@ -115,8 +125,8 @@ tasks:
       - uv run --group dbt dbt docs generate
       - uv run --group dbt dbt docs serve
 
-  # ----- Airflow -----
-  airflow:build:
+  # ----- Docker -----
+  docker:build:
     desc: Build the custom Airflow image
     deps: [dbt:manifest]
     cmds:
diff --git a/dags/repolytics_daily.py b/dags/repolytics_daily.py
index f9f273f..13b7ea2 100644
--- a/dags/repolytics_daily.py
+++ b/dags/repolytics_daily.py
@@ -1,8 +1,9 @@
 """Daily Repolytics pipeline: dlt ingestion -> dbt transform.
 
-Ingestion lands GitHub + PyPI into the DuckDB ``raw`` schema in a single atomic
-dlt load. Cosmos renders the dbt project, ``max_active_tasks=1`` serializes everything
-so the dbt model tasks never open the single-writer DuckDB file concurrently.
+Ingestion runs as two independent tasks - GitHub and PyPI. Either can fail/retry
+without touching the other. Both land into the DuckDB `raw` schema and are
+upstream of the dbt transform. Cosmos renders the dbt project; `max_active_tasks=1`
+serializes everything so no two tasks open the single-writer DuckDB file at once.
 """
 
 import os
@@ -38,11 +39,18 @@
 )
 def repolytics_daily():
     @task
-    def ingest() -> None:
-        """Run the dlt pipeline (GitHub + PyPI) into the DuckDB ``raw`` dataset."""
-        from repolytics.ingestion.pipeline import run
+    def ingest_github() -> None:
+        """Run GitHub ingestion into the DuckDB `raw` dataset."""
+        from repolytics.ingestion.pipeline import run_github
 
-        run()
+        run_github()
+
+    @task
+    def ingest_pypi() -> None:
+        """Run PyPI ingestion into the DuckDB `raw` dataset."""
+        from repolytics.ingestion.pipeline import run_pypi
+
+        run_pypi()
 
     transform = DbtTaskGroup(
         group_id="transform",
@@ -60,7 +68,7 @@ def ingest() -> None:
         ),
     )
 
-    ingest() >> transform
+    [ingest_github(), ingest_pypi()] >> transform
 
 
 repolytics_daily()
diff --git a/src/repolytics/ingestion/pipeline.py b/src/repolytics/ingestion/pipeline.py
index 7bcd718..d11ae1b 100644
--- a/src/repolytics/ingestion/pipeline.py
+++ b/src/repolytics/ingestion/pipeline.py
@@ -16,28 +16,44 @@ def build_pipeline(settings: Settings) -> dlt.Pipeline:
     )
 
 
-def run(settings: Settings | None = None) -> None:
-    """Run GitHub + PyPI ingestion into the configured DuckDB warehouse."""
+def run_github(settings: Settings | None = None) -> None:
+    """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.
+    """
     settings = settings or get_settings()
 
     repos = settings.target_repos
-    packages = settings.packages
-    if not repos and not packages:
-        raise RuntimeError(f"No projects to ingest - check {settings.projects_file}")
+    if not repos:
+        print(f"No repos to ingest - check {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))
+
 
-    # Run both sources in a single load so GitHub + PyPI land atomically.
-    sources = []
-    if repos:
-        sources.append(github_source(repos, settings.github_token.get_secret_value()))
-    if packages:
-        sources.append(pypi_source(packages))
+def run_pypi(settings: Settings | None = None) -> None:
+    """Run PyPI ingestion into the configured DuckDB warehouse.
+
+    No-ops (logs and returns) when no packages are configured.
+    """
+    settings = settings or get_settings()
 
-    info = pipeline.run(sources)
-    print(info)
+    packages = settings.packages
+    if not packages:
+        print(f"No packages to ingest - check {settings.projects_file}")
+        return
+
+    settings.duckdb_path.parent.mkdir(parents=True, exist_ok=True)
+    pipeline = build_pipeline(settings)
+    print(pipeline.run(pypi_source(packages)))
 
 
 if __name__ == "__main__":
-    run()
+    # Run GitHub + PyPI ingestion (CLI convenience wrapper for both sources).
+    settings = get_settings()
+    run_github(settings)
+    run_pypi(settings)