feat: migrate PyPI ingestion to BigQuery and add GitHub incremental loading

.env.example

@@ -1,7 +1,9 @@
 # GitHub
 GITHUB_TOKEN=ghp_xxxxxxxxxxxxxxxxxxxx
 
+# Google Cloud - PyPI download stats via the BigQuery public dataset
+GCP_PROJECT=your-gcp-project-id
+
 # Airflow
-AIRFLOW_UID=50000
-# Generate: uv run python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
-FERNET_KEY=secret
+AIRFLOW_UID=50000   # echo -e "AIRFLOW_UID=$(id -u)"
+FERNET_KEY=secret   # uv run python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

dags/repolytics_daily.py

@@ -1,14 +1,17 @@
 """Daily Repolytics pipeline: dlt ingestion -> dbt transform.
 
-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.
+Ingestion runs as two independent tasks. PyPI is date-windowed: it queries the
+BigQuery public dataset for the run's `data_interval_start`, so the DAG is
+backfillable per day. GitHub uses dlt incremental cursors (global, not per-interval),
+so it is skipped on historical backfill runs. Both land into the DuckDB `raw` schema
+upstream of the dbt transform. Cosmos renders the dbt project; `max_active_tasks=1` +
+`max_active_runs=1` serialize everything so no two tasks/runs open the single-writer
+DuckDB file at once.
 """
 
 import logging
 import os
-from datetime import timedelta
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from airflow.sdk import dag, task
@@ -50,7 +53,8 @@ def log_task_failure(context) -> None:
     dag_id="repolytics_daily",
     schedule="@daily",
     catchup=False,
-    max_active_tasks=1,  # DuckDB is single-writer: never run two dbt tasks at once.
+    max_active_tasks=1,  # DuckDB is single-writer: never run two dbt tasks at once
+    max_active_runs=1,  # serialize runs so a backfill never overlaps DuckDB writers
     default_args={
         "retries": 2,
         "retry_delay": timedelta(minutes=5),
@@ -61,23 +65,39 @@ def log_task_failure(context) -> None:
 def repolytics_daily():
     @task(multiple_outputs=False)
     def ingest_github() -> dict[str, int]:
-        """Run GitHub ingestion into the DuckDB `raw` dataset.
+        """Run GitHub incremental ingestion into the DuckDB `raw` dataset.
 
+        Skipped on historical backfill runs: the dlt cursor is global (not
+        per-interval), so replaying old intervals for GitHub is meaningless.
         Returns per-table row counts (pushed to XCom for the summary task).
         """
+        from airflow.sdk import get_current_context
+
         from repolytics.ingestion.pipeline import run_github
 
+        data_interval_end = get_current_context()["data_interval_end"]
+        if (datetime.now(UTC) - data_interval_end).days > 1:
+            logger.info(
+                "Skipping GitHub ingest for backfill interval ending %s",
+                data_interval_end,
+            )
+            return {}
         return run_github()
 
     @task(multiple_outputs=False)
     def ingest_pypi() -> dict[str, int]:
-        """Run PyPI ingestion into the DuckDB `raw` dataset.
+        """Run PyPI ingestion for the run's data-interval day (BigQuery).
 
-        Returns per-table row counts (pushed to XCom for the summary task).
+        Uses `data_interval_start`, so each run loads exactly one day and
+        the DAG backfills cleanly per interval. Returns per-table row counts
+        (pushed to XCom for the summary task).
         """
+        from airflow.sdk import get_current_context
+
         from repolytics.ingestion.pipeline import run_pypi
 
-        return run_pypi()
+        target_date = get_current_context()["data_interval_start"].date()
+        return run_pypi(target_date=target_date)
 
     transform = DbtTaskGroup(
         group_id="transform",

dbt/models/marts/_marts__models.yml

@@ -383,7 +383,7 @@ models:
 
   - name: fct_daily_downloads
     description: >
-      One row per package per day (PyPI 'without_mirrors' downloads). Incremental
+      One row per package per day (PyPI non-mirror downloads). Incremental
       (delete+insert on download_key).
     data_tests:
       - dbt_expectations.expect_table_row_count_to_be_between:

dbt/models/marts/fct_daily_downloads.sql

@@ -1,10 +1,8 @@
 -- PyPI daily download fact: one row per package per day. `repository_key` is resolved
 -- through the `projects` seed (package -> repo) and the SCD2 dim_repositories half-open
--- range; it is nullable for packages with no mapped/ingested repo.
--- Filtered to the 'without_mirrors' overall time-series category to avoid double
--- counting 'with_mirrors' and to exclude the recent-endpoint 'last_*' aggregates.
--- Incremental (delete+insert on download_key): only processes days at or after the
--- latest download_date already loaded; the unique key keeps it idempotent.
+-- range; it is nullable for packages with no mapped/ingested repo. Incremental
+-- (delete+insert on download_key): only processes days at or after the latest
+-- download_date already loaded; the unique key keeps it idempotent.
 
 {{
     config(
@@ -17,10 +15,9 @@
 
 with downloads as (
     select * from {{ ref('stg_pypi__downloads') }}
-    where category = 'without_mirrors'
     {% if is_incremental() %}
     -- date_key is the only date column on the target table; compare the day's key.
-    and {{ date_key('download_date') }} >= (select max(date_key) from {{ this }})
+    where {{ date_key('download_date') }} >= (select max(date_key) from {{ this }})
     {% endif %}
 )
 

dbt/models/staging/pypi/_pypi__models.yml

@@ -2,22 +2,15 @@ version: 2
 
 models:
   - name: stg_pypi__downloads
-    description: One cleaned row per package/category/day download count.
+    description: One cleaned row per package/day non-mirror download count.
     columns:
       - name: package
         data_tests: [not_null]
-      - name: category
-        description: PyPI overall time-series category.
-        data_tests:
-          - not_null
-          - accepted_values:
-              arguments:
-                values: [with_mirrors, without_mirrors]
       - name: download_date
         data_tests: [not_null]
       - name: download_count
         data_tests: [not_null]
     data_tests:
       - dbt_utils.unique_combination_of_columns:
           arguments:
-            combination_of_columns: [package, category, download_date]
+            combination_of_columns: [package, download_date]

dbt/models/staging/pypi/_pypi__sources.yml

@@ -2,7 +2,7 @@ version: 2
 
 sources:
   - name: pypi
-    description: PyPI Stats data loaded by dlt; one row per package/day/category.
+    description: PyPI download data loaded by dlt from BigQuery; one row per package/day.
     schema: raw
     loaded_at_field: _loaded_at
     freshness:

dbt/models/staging/pypi/stg_pypi__downloads.sql

@@ -6,7 +6,6 @@ with downloads as (
 
 select
     _package as package,
-    category,
     date::date as download_date,
     downloads as download_count,
     _loaded_at

docker-compose.yml

@@ -79,6 +79,8 @@ x-airflow-common:
     DUCKDB_PATH: /opt/airflow/data/warehouse/repolytics.duckdb
     # dbt project path
     DBT_PROJECT_DIR: /opt/airflow/dbt
+    # BigQuery (PyPI download stats): ADC file mounted from the host (see volumes).
+    GOOGLE_APPLICATION_CREDENTIALS: /opt/airflow/gcloud/adc.json
   volumes:
     - ${AIRFLOW_PROJ_DIR:-.}/dags:/opt/airflow/dags
     - ${AIRFLOW_PROJ_DIR:-.}/logs:/opt/airflow/logs
@@ -90,6 +92,8 @@ x-airflow-common:
     - ${AIRFLOW_PROJ_DIR:-.}/data:/opt/airflow/data
     # repolytics package source (src-layout)
     - ${AIRFLOW_PROJ_DIR:-.}/src:/opt/airflow/project/src
+    # Google Cloud ADC
+    - ${HOME}/.config/gcloud/application_default_credentials.json:/opt/airflow/gcloud/adc.json:ro
   user: "${AIRFLOW_UID:-50000}:0"
   depends_on:
     &airflow-common-depends-on

pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">=3.13"
 dependencies = [
     "dlt[duckdb]>=1.28.0",
     "duckdb>=1.5.3",
+    "google-cloud-bigquery>=3.42.1",
     "pydantic-settings>=2.14.1",
 ]
 

src/repolytics/config.py

@@ -26,6 +26,9 @@ class Settings(BaseSettings):
     # DuckDB
     duckdb_path: Path = Path("data/warehouse/repolytics.duckdb")
 
+    # Google Cloud project that BigQuery PyPI-download jobs are billed to.
+    gcp_project: str | None = None
+
     # Project list (repo <-> package) - also loaded by dbt as the `projects` seed.
     projects_file: Path = Path("dbt/seeds/projects.csv")
 

src/repolytics/ingestion/github_source.py

@@ -44,10 +44,19 @@ def repositories() -> Iterator[dict]:
             yield stamp(_repo=repo)(response.json())
 
     @dlt.resource(name="commits", write_disposition="merge", primary_key="sha")
-    def commits() -> Iterator[dict]:
+    def commits(
+        updated: dlt.sources.incremental[str] = dlt.sources.incremental(  # noqa: B008
+            "commit.author.date"
+        ),
+    ) -> Iterator[dict]:
+        # `since` (server-side, on commit date) bounds the fetch; dlt filters + merge
+        # keep it idempotent. The cursor is global across repos and commit dates can be
+        # backdated (rebases), so a backdated commit pushed today may be missed - the
+        # same trade-off the fct_commits watermark documents.
+        params = {"since": updated.last_value} if updated.last_value else {}
         for repo in repos:
             owner, name = repo.split("/")
-            yield from _paginate(f"/repos/{owner}/{name}/commits", repo)
+            yield from _paginate(f"/repos/{owner}/{name}/commits", repo, params)
 
     @dlt.resource(
         name="issues",
@@ -57,22 +66,38 @@ def commits() -> Iterator[dict]:
         # PR-shaped issues, so the staging PR filter never references a missing column.
         columns={"pull_request__url": {"data_type": "text", "nullable": True}},
     )
-    def issues() -> Iterator[dict]:
+    def issues(
+        updated: dlt.sources.incremental[str] = dlt.sources.incremental(  # noqa: B008
+            "updated_at"
+        ),
+    ) -> Iterator[dict]:
+        # `since` filters server-side on updated_at; sort asc so the cursor advances
+        # monotonically. Global cursor across repos (see commits caveat).
+        params = {"state": "all", "sort": "updated", "direction": "asc"}
+        if updated.last_value:
+            params["since"] = updated.last_value
         for repo in repos:
             owner, name = repo.split("/")
-            yield from _paginate(
-                f"/repos/{owner}/{name}/issues", repo, {"state": "all"}
-            )
+            yield from _paginate(f"/repos/{owner}/{name}/issues", repo, params)
 
     @dlt.resource(
         name="pull_requests",
         write_disposition="merge",
         primary_key=["_repo", "number"],  # PR number is unique per repo
     )
-    def pull_requests() -> Iterator[dict]:
+    def pull_requests(
+        updated: dlt.sources.incremental[str] = dlt.sources.incremental(  # noqa: B008
+            "updated_at"
+        ),
+    ) -> Iterator[dict]:
+        # The /pulls endpoint has no `since`, so we can't bound the fetch server-side
+        # and can't use dlt's row_order early-exit either (it assumes one monotonic
+        # stream, but looping repos makes the updated_at sequence saw-tooth). We page
+        # the PR list and let dlt's cursor + merge drop/upsert unchanged rows.
+        params = {"state": "all", "sort": "updated", "direction": "desc"}
         for repo in repos:
             owner, name = repo.split("/")
-            yield from _paginate(f"/repos/{owner}/{name}/pulls", repo, {"state": "all"})
+            yield from _paginate(f"/repos/{owner}/{name}/pulls", repo, params)
 
     @dlt.resource(name="releases", write_disposition="merge", primary_key="id")
     def releases() -> Iterator[dict]:

src/repolytics/ingestion/pipeline.py

@@ -1,6 +1,7 @@
 """dlt pipelines wiring GitHub and PyPI ingestion into the DuckDB `raw` dataset."""
 
 import logging
+from datetime import UTC, date, datetime, timedelta
 
 import dlt
 
@@ -51,11 +52,14 @@ def run_github(settings: Settings | None = None) -> dict[str, int]:
     return _table_row_counts(pipeline)
 
 
-def run_pypi(settings: Settings | None = None) -> dict[str, int]:
-    """Run PyPI ingestion into the configured DuckDB warehouse.
+def run_pypi(
+    settings: Settings | None = None, target_date: date | None = None
+) -> dict[str, int]:
+    """Run PyPI ingestion for a single day into the configured DuckDB warehouse.
 
-    No-ops (logs and returns `{}`) when no packages are configured. Returns the
-    per-table row counts loaded in this run.
+    Queries the BigQuery public dataset for `target_date` (defaults to yesterday
+    UTC, the most recent complete partition). 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()
 
@@ -64,9 +68,11 @@ def run_pypi(settings: Settings | None = None) -> dict[str, int]:
         logger.warning("No packages to ingest - check %s", settings.projects_file)
         return {}
 
+    target_date = target_date or (datetime.now(UTC).date() - timedelta(days=1))
     settings.duckdb_path.parent.mkdir(parents=True, exist_ok=True)
     pipeline = build_pipeline(settings)
-    logger.info("PyPI ingestion complete: %s", pipeline.run(pypi_source(packages)))
+    source = pypi_source(packages, target_date, project=settings.gcp_project)
+    logger.info("PyPI ingestion complete (%s): %s", target_date, pipeline.run(source))
     return _table_row_counts(pipeline)