Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions Taskfile.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
24 changes: 16 additions & 8 deletions dags/repolytics_daily.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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",
Expand All @@ -60,7 +68,7 @@ def ingest() -> None:
),
)

ingest() >> transform
[ingest_github(), ingest_pypi()] >> transform


repolytics_daily()
44 changes: 30 additions & 14 deletions src/repolytics/ingestion/pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)