refactor: replace custom ingestion/loading with dlt

.env.example

@@ -1,14 +1,9 @@
 # GitHub
 GITHUB_TOKEN=ghp_xxxxxxxxxxxxxxxxxxxx
-GITHUB_TARGET_REPOS=fastapi/fastapi,pydantic/pydantic,pola-rs/polars,duckdb/duckdb,encode/httpx
 
 # DuckDB
 DUCKDB_PATH=data/warehouse/repolytics.duckdb
 
-# Paths
-RAW_DATA_PATH=data/raw
-WATERMARKS_PATH=data/raw/.watermarks.json
-
 # Airflow
 AIRFLOW_HOME=orchestration
 AIRFLOW__CORE__DAGS_FOLDER=orchestration/dags

.gitignore

@@ -21,11 +21,14 @@ venv/
 .coverage.*
 htmlcov/
 
-# Runtime data (DuckDB warehouse, raw Parquet landing zone)
+# Runtime data (DuckDB warehouse)
 data/
 *.duckdb
 *.duckdb.wal
 
+# dlt (local config/secrets + pipeline working dir)
+.dlt/
+
 # Logs
 logs/
 orchestration/logs/

Taskfile.yml

@@ -14,6 +14,11 @@ tasks:
     cmds:
       - uv run ruff check .
 
+  lint:fix:
+    desc: Run ruff linter (applies fixes)
+    cmds:
+      - uv run ruff check --fix .
+
   format:
     desc: Run ruff formatter (modifies files)
     cmds:
@@ -54,13 +59,19 @@ tasks:
       - task: dbt:deps
       - task: dbt:compile
 
-  # ----- Loading -----
-  load:raw:
-    desc: Load landed Parquet into the DuckDB raw schema
+  # ----- dlt -----
+  dlt:ingest:
+    desc: Run the dlt pipeline (GitHub + PyPI) into the DuckDB raw schema
     cmds:
-      - uv run python -c "from repolytics.loading.raw_loader import load_all; print(load_all())"
+      - uv run python -m repolytics.ingestion.pipeline
 
   # ----- dbt -----
+  dbt:seed:
+    desc: Load CSV seeds into the warehouse
+    dir: dbt
+    cmds:
+      - uv run --group dbt dbt seed
+
   dbt:deps:
     desc: Install dbt packages
     dir: dbt

dbt/macros/date_key.sql

@@ -0,0 +1,7 @@
+{#-
+  Surrogate date key (YYYYMMDD integer) from a timestamp/date expression.
+  Centralizes the format used by every fact join to dim_dates.
+-#}
+{% macro date_key(ts) -%}
+    cast(strftime({{ ts }}, '%Y%m%d') as integer)
+{%- endmacro %}

dbt/macros/scd2_repository_join.sql

@@ -0,0 +1,18 @@
+{#-
+  Resolve a fact repository_key against the SCD2 dim_repositories using a
+  half-open date range (event_date >= valid_from AND < valid_to), so each event
+  maps to the repository version that was current when it occurred. Centralizes
+  the SCD2 join contract (half-open bounds; earliest version backdated, current
+  version valid_to = 9999-12-31) that every fact shares. Select
+  `<alias>.repository_key` after using this.
+
+    repo_expr - SQL expression for the repository full name (owner/name)
+    date_expr - SQL expression for the event date (cast to date)
+    alias     - alias bound to dim_repositories (default 'r')
+-#}
+{% macro scd2_repository_join(repo_expr, date_expr, alias='r') -%}
+left join {{ ref('dim_repositories') }} {{ alias }}
+    on {{ repo_expr }} = {{ alias }}.repository_name
+    and {{ date_expr }} >= {{ alias }}.valid_from
+    and {{ date_expr }} < {{ alias }}.valid_to
+{%- endmacro %}

dbt/macros/source_or_empty.sql

@@ -0,0 +1,26 @@
+{#-
+  Return `select * from <source>` when the source relation exists, otherwise an
+  empty result with the given columns (`select cast(null as <type>) as <col> ... where false`).
+
+  dlt only materializes a normalized child table when some row populated the
+  underlying list, so a list that is empty across every ingested row leaves the
+  child table absent. Wrapping the source in this macro keeps the build resilient
+  to that case.
+
+  `columns` is a mapping of column name -> SQL type, covering the columns the caller reads.
+-#}
+{% macro source_or_empty(source_name, table_name, columns) -%}
+    {%- set rel = source(source_name, table_name) -%}
+    {%- set existing = adapter.get_relation(
+        database=rel.database, schema=rel.schema, identifier=rel.identifier
+    ) -%}
+    {%- if existing is not none -%}
+        select * from {{ rel }}
+    {%- else -%}
+        select
+        {%- for col, dtype in columns.items() %}
+            cast(null as {{ dtype }}) as {{ col }}{{ "," if not loop.last }}
+        {%- endfor %}
+        where false
+    {%- endif -%}
+{%- endmacro %}

dbt/models/marts/_marts__models.yml

@@ -71,16 +71,6 @@ models:
               arguments:
                 to: ref('dim_dates')
                 field: date_key
-      - name: lines_added
-        data_tests:
-          - dbt_utils.expression_is_true:
-              arguments:
-                expression: ">= 0"
-      - name: lines_deleted
-        data_tests:
-          - dbt_utils.expression_is_true:
-              arguments:
-                expression: ">= 0"
 
   - name: fct_pull_requests
     description: One row per pull request. Repository resolved via the SCD2 half-open range on the PR open date.
@@ -116,11 +106,6 @@ models:
               arguments:
                 to: ref('dim_dates')
                 field: date_key
-      - name: review_count
-        data_tests:
-          - dbt_utils.expression_is_true:
-              arguments:
-                expression: ">= 0"
 
   - name: fct_issues
     description: One row per issue. Repository resolved via the SCD2 half-open range.
@@ -186,15 +171,23 @@ models:
 
   - name: fct_daily_downloads
     description: >
-      One row per package per day (PyPI 'without_mirrors' downloads). `package` is a
-      degenerate dimension - no package->repository mapping exists yet. Incremental
+      One row per package per day (PyPI 'without_mirrors' downloads). Incremental
       (delete+insert on download_key).
     columns:
       - name: download_key
         description: Surrogate key, hash of package + download_date.
         data_tests: [not_null, unique]
       - name: package
         description: PyPI package name (degenerate dimension).
+      - name: repository_key
+        description: >
+          FK to dim_repositories via the projects (package->repo) seed; nullable when
+          the package has no mapped/ingested repository.
+        data_tests:
+          - relationships:
+              arguments:
+                to: ref('dim_repositories')
+                field: repository_key
       - name: date_key
         description: FK to dim_dates (download day, YYYYMMDD).
         data_tests:

dbt/models/marts/bridge_issue_labels.sql

@@ -1,27 +1,15 @@
--- Bridge table for the issue<->label many-to-many. Labels land in staging as a JSON
--- array per issue; we explode them, recompute the issue surrogate key the same way as
--- fct_issues, and resolve each label name to its label_key in dim_labels.
+-- Bridge table for the issue<->label many-to-many. Built from the flattened issue
+-- label staging model; recomputes the issue surrogate key the same way as fct_issues
+-- and resolves each label name to its label_key in dim_labels.
 -- Grain: one row per (issue, label).
 
 with issue_labels as (
-    select
-        repository,
-        issue_number,
-        unnest(json_extract(labels, '$[*]')) as label
-    from {{ ref('stg_github__issues') }}
-    where labels is not null
-),
-
-exploded as (
-    select
-        {{ dbt_utils.generate_surrogate_key(['repository', 'issue_number']) }} as issue_key,
-        label ->> '$.name' as label_name
-    from issue_labels
-    where (label ->> '$.name') is not null
+    select * from {{ ref('stg_github__issue_labels') }}
 )
 
 select distinct
-    e.issue_key,
+    {{ dbt_utils.generate_surrogate_key(['il.repository', 'il.issue_number']) }}
+        as issue_key,
     dl.label_key
-from exploded e
-inner join {{ ref('dim_labels') }} dl on e.label_name = dl.label_name
+from issue_labels il
+inner join {{ ref('dim_labels') }} dl on il.label_name = dl.label_name

dbt/models/marts/dim_labels.sql

@@ -1,28 +1,19 @@
--- Type 1 label dimension: distinct issue/PR labels across all projects. Labels land
--- in staging as a JSON array per row, so we explode the arrays and dedupe by name.
+-- Type 1 label dimension: distinct issue/PR labels across all projects. Labels come
+-- from the flattened dlt child-table staging models, deduped by name.
 
-with labels_raw as (
-    select labels
-    from {{ ref('stg_github__pull_requests') }}
-    where labels is not null
+with labels as (
+    select label_name, label_color from {{ ref('stg_github__issue_labels') }}
     union all
-    select labels
-    from {{ ref('stg_github__issues') }}
-    where labels is not null
-),
-
-exploded as (
-    select unnest(json_extract(labels, '$[*]')) as label
-    from labels_raw
+    select label_name, label_color from {{ ref('stg_github__pr_labels') }}
 ),
 
 distinct_labels as (
     select
-        label ->> '$.name' as label_name,
-        max(label ->> '$.color') as label_color
-    from exploded
-    where (label ->> '$.name') is not null
-    group by label ->> '$.name'
+        label_name,
+        max(label_color) as label_color
+    from labels
+    where label_name is not null
+    group by label_name
 )
 
 select

dbt/models/marts/fct_commits.sql

@@ -2,8 +2,10 @@
 -- half-open date range (event_date >= valid_from AND < valid_to) so each commit maps
 -- to the repository version that was current when it landed; to dim_contributors on
 -- author login; and to dim_dates via an inline YYYYMMDD key.
--- Incremental (delete+insert on commit_key): each run only processes commits at or
--- after the latest committed_at already loaded; the unique key keeps it idempotent.
+-- Incremental (delete+insert on commit_key): each run only processes rows loaded
+-- since the last run, keyed on the monotonic dlt `_loaded_at` (the git author date is
+-- not monotonic, so a rebased/old-authored commit pushed today would be skipped).
+-- The unique key keeps it idempotent.
 
 {{
     config(
@@ -17,23 +19,19 @@
 with commits as (
     select * from {{ ref('stg_github__commits') }}
     {% if is_incremental() %}
-    where committed_at >= (select max(committed_at) from {{ this }})
+    where _loaded_at >= (select max(_loaded_at) from {{ this }})
     {% endif %}
 )
 
 select
     {{ dbt_utils.generate_surrogate_key(['c.repository', 'c.commit_sha']) }} as commit_key,
     r.repository_key,
     co.contributor_key,
-    cast(strftime(c.committed_at, '%Y%m%d') as integer) as date_key,
+    {{ date_key('c.committed_at') }} as date_key,
     c.commit_sha as commit_hash,
-    c.additions as lines_added,
-    c.deletions as lines_deleted,
-    c.committed_at
+    c.committed_at,
+    c._loaded_at
 from commits c
-left join {{ ref('dim_repositories') }} r
-    on c.repository = r.repository_name
-    and c.committed_at::date >= r.valid_from
-    and c.committed_at::date < r.valid_to
+{{ scd2_repository_join('c.repository', 'c.committed_at::date') }}
 left join {{ ref('dim_contributors') }} co
     on c.author_login = co.username

dbt/models/marts/fct_contributor_activity_monthly.sql

@@ -21,7 +21,7 @@ select
     c.contributor_key,
     m.username,
     m.event_month,
-    cast(strftime(m.event_month, '%Y%m%d') as integer) as month_date_key,
+    {{ date_key('m.event_month') }} as month_date_key,
     m.commits,
     m.prs_opened,
     m.prs_merged,

dbt/models/marts/fct_daily_downloads.sql

@@ -1,4 +1,6 @@
--- PyPI daily download fact: one row per package per day.
+-- 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
@@ -18,14 +20,17 @@ with downloads as (
     where category = 'without_mirrors'
     {% if is_incremental() %}
     -- date_key is the only date column on the target table; compare the day's key.
-    and cast(strftime(download_date, '%Y%m%d') as integer)
-        >= (select max(date_key) from {{ this }})
+    and {{ date_key('download_date') }} >= (select max(date_key) from {{ this }})
     {% endif %}
 )
 
 select
-    {{ dbt_utils.generate_surrogate_key(['package', 'download_date']) }} as download_key,
-    package,
-    cast(strftime(download_date, '%Y%m%d') as integer) as date_key,
-    download_count
-from downloads
+    {{ dbt_utils.generate_surrogate_key(['d.package', 'd.download_date']) }}
+        as download_key,
+    d.package,
+    r.repository_key,
+    {{ date_key('d.download_date') }} as date_key,
+    d.download_count
+from downloads d
+left join {{ ref('projects') }} p on d.package = p.package
+{{ scd2_repository_join('p.repo', 'd.download_date') }}

dbt/models/marts/fct_issues.sql

@@ -10,15 +10,15 @@ select
     {{ dbt_utils.generate_surrogate_key(['i.repository', 'i.issue_number']) }} as issue_key,
     r.repository_key,
     co.contributor_key as author_key,
-    cast(strftime(i.created_at, '%Y%m%d') as integer) as opened_date_key,
-    cast(strftime(i.closed_at, '%Y%m%d') as integer) as closed_date_key,
-    i.is_closed,
-    i.time_to_close_hours,
+    {{ date_key('i.created_at') }} as opened_date_key,
+    {{ date_key('i.closed_at') }} as closed_date_key,
+    i.state = 'closed' as is_closed,
+    case
+        when i.closed_at is not null
+            then datediff('hour', i.created_at, i.closed_at)
+    end as time_to_close_hours,
     i.comment_count
 from issues i
-left join {{ ref('dim_repositories') }} r
-    on i.repository = r.repository_name
-    and i.created_at::date >= r.valid_from
-    and i.created_at::date < r.valid_to
+{{ scd2_repository_join('i.repository', 'i.created_at::date') }}
 left join {{ ref('dim_contributors') }} co
     on i.author_login = co.username

dbt/models/marts/fct_pull_requests.sql

@@ -10,18 +10,14 @@ select
     {{ dbt_utils.generate_surrogate_key(['p.repository', 'p.pr_number']) }} as pr_key,
     r.repository_key,
     co.contributor_key as author_key,
-    cast(strftime(p.created_at, '%Y%m%d') as integer) as opened_date_key,
-    cast(strftime(p.merged_at, '%Y%m%d') as integer) as merged_date_key,
+    {{ date_key('p.created_at') }} as opened_date_key,
+    {{ date_key('p.merged_at') }} as merged_date_key,
     p.merged_at is not null as is_merged,
-    p.time_to_merge_hours,
-    p.review_comments as review_count,
-    p.comment_count,
-    p.additions,
-    p.deletions
+    case
+        when p.merged_at is not null
+            then datediff('hour', p.created_at, p.merged_at)
+    end as time_to_merge_hours
 from pull_requests p
-left join {{ ref('dim_repositories') }} r
-    on p.repository = r.repository_name
-    and p.created_at::date >= r.valid_from
-    and p.created_at::date < r.valid_to
+{{ scd2_repository_join('p.repository', 'p.created_at::date') }}
 left join {{ ref('dim_contributors') }} co
     on p.author_login = co.username