feat: setup Airflow with Cosmos and orchestrate the ELT pipeline

.env.example

@@ -1,14 +1,7 @@
 # GitHub
 GITHUB_TOKEN=ghp_xxxxxxxxxxxxxxxxxxxx
 
-# DuckDB
-DUCKDB_PATH=data/warehouse/repolytics.duckdb
-
 # Airflow
-AIRFLOW_HOME=orchestration
-AIRFLOW__CORE__DAGS_FOLDER=orchestration/dags
-AIRFLOW__CORE__LOAD_EXAMPLES=false
-
-# dbt
-DBT_PROJECT_DIR=dbt
-DBT_PROFILES_DIR=dbt
+AIRFLOW_UID=50000
+# Generate: uv run python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
+FERNET_KEY=secret

.gitignore

@@ -26,19 +26,17 @@ data/
 *.duckdb
 *.duckdb.wal
 
-# dlt (local config/secrets + pipeline working dir)
-.dlt/
-
-# Logs
-logs/
-orchestration/logs/
+# Airflow
+config/airflow.cfg
 
 # dbt
 dbt/target/
 dbt/dbt_packages/
-dbt/logs/
 dbt/.user.yml
 
+# Logs
+logs/
+
 # OS / editor
 .DS_Store
 Thumbs.db

Dockerfile

@@ -0,0 +1,29 @@
+# Custom Airflow image: extends the official image with Cosmos + dbt, installed
+# from uv.lock so the runtime matches the local .venv.
+
+FROM apache/airflow:3.2.2-python3.13
+
+COPY --from=ghcr.io/astral-sh/uv:0.11.23 /uv /uvx /bin/
+
+ENV UV_LINK_MODE=copy
+
+# Install as root: the system site-packages (/usr/python) is root-owned
+USER root
+
+# Resolve the orchestration + dbt groups from the lockfile and install them into
+# the image's system Python. --no-emit-package apache-airflow excludes the meta
+# package so the base image's Airflow is left untouched.
+COPY pyproject.toml uv.lock /tmp/build/
+RUN uv export --frozen --no-emit-project --no-hashes \
+      --directory /tmp/build \
+      --no-emit-package apache-airflow \
+      --group orchestration --group dbt \
+    | uv pip install --system --no-cache -r - \
+ && rm -rf /tmp/build
+
+# Install the repolytics package, --no-deps because deps already installed above.
+COPY --chown=airflow:0 pyproject.toml uv.lock README.md /opt/airflow/project/
+COPY --chown=airflow:0 src /opt/airflow/project/src
+RUN uv pip install --system --no-cache --no-deps -e /opt/airflow/project
+
+USER airflow

Taskfile.yml

@@ -84,6 +84,12 @@ tasks:
     cmds:
       - uv run --group dbt dbt compile --target ci
 
+  dbt:manifest:
+    desc: Generate target/manifest.json for Cosmos (DBT_MANIFEST load mode)
+    dir: dbt
+    cmds:
+      - uv run --group dbt dbt parse
+
   dbt:run:
     desc: Run dbt models
     dir: dbt
@@ -108,3 +114,10 @@ tasks:
     cmds:
       - uv run --group dbt dbt docs generate
       - uv run --group dbt dbt docs serve
+
+  # ----- Airflow -----
+  airflow:build:
+    desc: Build the custom Airflow image
+    deps: [dbt:manifest]
+    cmds:
+      - docker compose build

dags/repolytics_daily.py

@@ -0,0 +1,66 @@
+"""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.
+"""
+
+import os
+from datetime import timedelta
+from pathlib import Path
+
+from airflow.sdk import dag, task
+from cosmos import (
+    DbtTaskGroup,
+    LoadMode,
+    ProfileConfig,
+    ProjectConfig,
+    RenderConfig,
+)
+
+# Location of the dbt project inside the container
+DBT_PROJECT_DIR = Path(os.environ.get("DBT_PROJECT_DIR", "/opt/airflow/dbt"))
+
+profile_config = ProfileConfig(
+    profile_name="repolytics",
+    target_name="dev",
+    profiles_yml_filepath=DBT_PROJECT_DIR / "profiles.yml",
+)
+
+
+@dag(
+    dag_id="repolytics_daily",
+    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)},
+    tags=["repolytics", "elt"],
+)
+def repolytics_daily():
+    @task
+    def ingest() -> None:
+        """Run the dlt pipeline (GitHub + PyPI) into the DuckDB ``raw`` dataset."""
+        from repolytics.ingestion.pipeline import run
+
+        run()
+
+    transform = DbtTaskGroup(
+        group_id="transform",
+        project_config=ProjectConfig(
+            dbt_project_path=DBT_PROJECT_DIR,
+            # Render the graph from a pre-built manifest.
+            manifest_path=DBT_PROJECT_DIR / "target" / "manifest.json",
+        ),
+        profile_config=profile_config,
+        render_config=RenderConfig(
+            load_method=LoadMode.DBT_MANIFEST,
+            # Detach tests that depend on multiple models into their own
+            # tasks scheduled after all parents are built.
+            should_detach_multiple_parents_tests=True,
+        ),
+    )
+
+    ingest() >> transform
+
+
+repolytics_daily()

dbt/dbt_project.yml

@@ -29,6 +29,10 @@ models:
       +materialized: table
       +schema: marts
 
+seeds:
+  repolytics:
+    +schema: staging
+
 snapshots:
   repolytics:
     +schema: snapshots

dbt/models/marts/dim_dates.sql

@@ -2,7 +2,7 @@ with spine as (
     {{
         dbt_utils.date_spine(
             datepart="day",
-            start_date="cast('2020-01-01' as date)",
+            start_date="cast('2000-01-01' as date)",
             end_date="cast('2031-01-01' as date)"
         )
     }}

dbt/seeds/projects.csv

@@ -1,6 +1,3 @@
 repo,package
 fastapi/fastapi,fastapi
 pydantic/pydantic,pydantic
-pola-rs/polars,polars
-duckdb/duckdb,duckdb
-encode/httpx,httpx