From f2832d8647ac8d7af18ab853183edbd77d12467d Mon Sep 17 00:00:00 2001 From: Neukz Date: Mon, 15 Jun 2026 17:00:34 +0200 Subject: [PATCH] feat: add raw loader and dbt staging models with tests --- .github/workflows/ci.yml | 13 +- .gitignore | 1 + Taskfile.yml | 40 ++++++ dbt/.gitkeep | 0 dbt/dbt_project.yml | 24 ++++ dbt/macros/generate_schema_name.sql | 12 ++ dbt/models/staging/github/_github__models.yml | 55 ++++++++ .../staging/github/_github__sources.yml | 21 +++ .../staging/github/stg_github__commits.sql | 19 +++ .../staging/github/stg_github__issues.sql | 26 ++++ .../github/stg_github__pull_requests.sql | 26 ++++ .../staging/github/stg_github__releases.sql | 13 ++ .../github/stg_github__repositories.sql | 21 +++ dbt/models/staging/pypi/_pypi__models.yml | 16 +++ dbt/models/staging/pypi/_pypi__sources.yml | 13 ++ .../staging/pypi/stg_pypi__downloads.sql | 12 ++ dbt/package-lock.yml | 5 + dbt/packages.yml | 3 + dbt/profiles.yml | 12 ++ src/repolytics/loading/raw_loader.py | 54 ++++++++ tests/fixtures/.gitkeep | 0 tests/fixtures/github_responses/commits.json | 20 +++ tests/fixtures/github_responses/issues.json | 23 ++++ tests/fixtures/github_responses/pulls.json | 30 +++++ tests/fixtures/github_responses/releases.json | 16 +++ .../fixtures/github_responses/repository.json | 15 +++ tests/fixtures/pypi_responses/overall.json | 8 ++ tests/fixtures/pypi_responses/recent.json | 5 + tests/integration/test_raw_loader.py | 121 ++++++++++++++++++ 29 files changed, 623 insertions(+), 1 deletion(-) delete mode 100644 dbt/.gitkeep create mode 100644 dbt/dbt_project.yml create mode 100644 dbt/macros/generate_schema_name.sql create mode 100644 dbt/models/staging/github/_github__models.yml create mode 100644 dbt/models/staging/github/_github__sources.yml create mode 100644 dbt/models/staging/github/stg_github__commits.sql create mode 100644 dbt/models/staging/github/stg_github__issues.sql create mode 100644 dbt/models/staging/github/stg_github__pull_requests.sql create mode 100644 dbt/models/staging/github/stg_github__releases.sql create mode 100644 dbt/models/staging/github/stg_github__repositories.sql create mode 100644 dbt/models/staging/pypi/_pypi__models.yml create mode 100644 dbt/models/staging/pypi/_pypi__sources.yml create mode 100644 dbt/models/staging/pypi/stg_pypi__downloads.sql create mode 100644 dbt/package-lock.yml create mode 100644 dbt/packages.yml create mode 100644 dbt/profiles.yml create mode 100644 src/repolytics/loading/raw_loader.py delete mode 100644 tests/fixtures/.gitkeep create mode 100644 tests/fixtures/github_responses/commits.json create mode 100644 tests/fixtures/github_responses/issues.json create mode 100644 tests/fixtures/github_responses/pulls.json create mode 100644 tests/fixtures/github_responses/releases.json create mode 100644 tests/fixtures/github_responses/repository.json create mode 100644 tests/fixtures/pypi_responses/overall.json create mode 100644 tests/fixtures/pypi_responses/recent.json create mode 100644 tests/integration/test_raw_loader.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index df24a08..9f0a113 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,7 +25,7 @@ jobs: cache-dependency-glob: uv.lock - name: Install dependencies - run: uv sync --locked + run: uv sync --locked --group dbt - name: Lint run: uv run --no-sync ruff check . @@ -35,3 +35,14 @@ jobs: - name: Unit tests run: uv run --no-sync pytest tests/unit + + - name: Integration tests + run: uv run --no-sync pytest tests/integration + + - name: Install dbt packages + working-directory: dbt + run: uv run --no-sync dbt deps + + - name: Compile dbt models + working-directory: dbt + run: uv run --no-sync dbt compile --target ci diff --git a/.gitignore b/.gitignore index 739c221..d1e9704 100644 --- a/.gitignore +++ b/.gitignore @@ -34,6 +34,7 @@ orchestration/logs/ dbt/target/ dbt/dbt_packages/ dbt/logs/ +dbt/.user.yml # OS / editor .DS_Store diff --git a/Taskfile.yml b/Taskfile.yml index 104fad3..57556db 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -50,3 +50,43 @@ tasks: - task: lint - task: format:check - task: test:unit + - task: test:integration + - task: dbt:deps + - task: dbt:compile + + # ----- Loading ----- + load:raw: + desc: Load landed Parquet into the DuckDB raw schema + cmds: + - uv run python -c "from repolytics.loading.raw_loader import load_all; print(load_all())" + + # ----- dbt ----- + dbt:deps: + desc: Install dbt packages + dir: dbt + cmds: + - uv run --group dbt dbt deps + + dbt:compile: + desc: Compile dbt models (validate SQL without executing) + dir: dbt + cmds: + - uv run --group dbt dbt compile --target ci + + dbt:run: + desc: Run dbt models + dir: dbt + cmds: + - uv run --group dbt dbt run + + dbt:test: + desc: Run dbt tests + dir: dbt + cmds: + - uv run --group dbt dbt test + + dbt:build: + desc: Run dbt models and tests + dir: dbt + cmds: + - uv run --group dbt dbt build diff --git a/dbt/.gitkeep b/dbt/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/dbt/dbt_project.yml b/dbt/dbt_project.yml new file mode 100644 index 0000000..c0ef196 --- /dev/null +++ b/dbt/dbt_project.yml @@ -0,0 +1,24 @@ +name: 'repolytics' +version: '1.0.0' + +config-version: 2 +profile: 'repolytics' + +require-dbt-version: [">=1.11.0", "<2.0.0"] + +model-paths: ["models"] +test-paths: ["tests"] +seed-paths: ["seeds"] +macro-paths: ["macros"] +snapshot-paths: ["snapshots"] + +target-path: target +clean-targets: + - "target" + - "dbt_packages" + +models: + repolytics: + staging: + +materialized: view + +schema: staging diff --git a/dbt/macros/generate_schema_name.sql b/dbt/macros/generate_schema_name.sql new file mode 100644 index 0000000..2152d30 --- /dev/null +++ b/dbt/macros/generate_schema_name.sql @@ -0,0 +1,12 @@ +{# + Use the configured custom schema name verbatim (e.g. `staging`, `marts`) + instead of dbt default `{target_schema}_{custom}` concatenation, so the + warehouse layers map to clean schema names. +#} +{% macro generate_schema_name(custom_schema_name, node) -%} + {%- if custom_schema_name is none -%} + {{ target.schema }} + {%- else -%} + {{ custom_schema_name | trim }} + {%- endif -%} +{%- endmacro %} diff --git a/dbt/models/staging/github/_github__models.yml b/dbt/models/staging/github/_github__models.yml new file mode 100644 index 0000000..0a1f61e --- /dev/null +++ b/dbt/models/staging/github/_github__models.yml @@ -0,0 +1,55 @@ +version: 2 + +models: + - name: stg_github__repositories + description: One cleaned row per repository snapshot. + columns: + - name: repository_id + description: Natural key from the GitHub API. + data_tests: [not_null, unique] + - name: repository_name + data_tests: [not_null] + + - name: stg_github__commits + description: One cleaned row per non-merge commit. + columns: + - name: commit_sha + data_tests: [not_null] + - name: repository + data_tests: [not_null] + data_tests: + - dbt_utils.unique_combination_of_columns: + arguments: + combination_of_columns: [repository, commit_sha] + + - name: stg_github__pull_requests + description: One cleaned row per pull request. + columns: + - name: pr_number + data_tests: [not_null] + - name: repository + data_tests: [not_null] + data_tests: + - dbt_utils.unique_combination_of_columns: + arguments: + combination_of_columns: [repository, pr_number] + + - name: stg_github__issues + description: One cleaned row per issue (pull requests excluded). + columns: + - name: issue_number + data_tests: [not_null] + - name: repository + data_tests: [not_null] + data_tests: + - dbt_utils.unique_combination_of_columns: + arguments: + combination_of_columns: [repository, issue_number] + + - name: stg_github__releases + description: One cleaned row per release. + columns: + - name: release_id + data_tests: [not_null, unique] + - name: repository + data_tests: [not_null] diff --git a/dbt/models/staging/github/_github__sources.yml b/dbt/models/staging/github/_github__sources.yml new file mode 100644 index 0000000..c17b8df --- /dev/null +++ b/dbt/models/staging/github/_github__sources.yml @@ -0,0 +1,21 @@ +version: 2 + +sources: + - name: github + description: Raw GitHub API responses, landed as a JSON `data` blob per record. + schema: raw + loaded_at_field: _loaded_at + freshness: + warn_after: {count: 12, period: hour} + error_after: {count: 24, period: hour} + tables: + - name: repositories + description: One row per repository snapshot. + - name: commits + description: One row per commit; `_repo` carries the source repository. + - name: pull_requests + description: One row per pull request; `_repo` carries the source repository. + - name: issues + description: One row per issue (includes PRs upstream; staging filters them). + - name: releases + description: One row per release; `_repo` carries the source repository. diff --git a/dbt/models/staging/github/stg_github__commits.sql b/dbt/models/staging/github/stg_github__commits.sql new file mode 100644 index 0000000..34d443c --- /dev/null +++ b/dbt/models/staging/github/stg_github__commits.sql @@ -0,0 +1,19 @@ +with source as ( + select data::json as d, _repo, _loaded_at + from {{ source('github', 'commits') }} +) + +select + d ->> '$.sha' as commit_sha, + _repo as repository, + d ->> '$.author.login' as author_login, + d ->> '$.commit.author.name' as author_name, + d ->> '$.commit.author.email' as author_email, + (d ->> '$.commit.author.date')::timestamp as committed_at, + (d ->> '$.stats.additions')::bigint as additions, + (d ->> '$.stats.deletions')::bigint as deletions, + d ->> '$.commit.message' as message, + _loaded_at +from source +-- Drop merge commits (more than one parent); rows without `parents` are kept. +where coalesce(json_array_length(d -> '$.parents'), 0) <= 1 diff --git a/dbt/models/staging/github/stg_github__issues.sql b/dbt/models/staging/github/stg_github__issues.sql new file mode 100644 index 0000000..2e10e86 --- /dev/null +++ b/dbt/models/staging/github/stg_github__issues.sql @@ -0,0 +1,26 @@ +with source as ( + select data::json as d, _repo, _loaded_at + from {{ source('github', 'issues') }} +) + +select + _repo as repository, + (d ->> '$.number')::bigint as issue_number, + d ->> '$.user.login' as author_login, + d ->> '$.state' as state, + (d ->> '$.state') = 'closed' as is_closed, + (d ->> '$.created_at')::timestamp as created_at, + (d ->> '$.closed_at')::timestamp as closed_at, + case + when (d ->> '$.closed_at') is not null then datediff( + 'hour', + (d ->> '$.created_at')::timestamp, + (d ->> '$.closed_at')::timestamp + ) + end as time_to_close_hours, + (d ->> '$.comments')::bigint as comment_count, + d -> '$.labels' as labels, + _loaded_at +from source +-- The issues endpoint returns PRs too; drop them (real issues have no `pull_request`). +where (d -> '$.pull_request') is null diff --git a/dbt/models/staging/github/stg_github__pull_requests.sql b/dbt/models/staging/github/stg_github__pull_requests.sql new file mode 100644 index 0000000..57757ba --- /dev/null +++ b/dbt/models/staging/github/stg_github__pull_requests.sql @@ -0,0 +1,26 @@ +with source as ( + select data::json as d, _repo, _loaded_at + from {{ source('github', 'pull_requests') }} +) + +select + _repo as repository, + (d ->> '$.number')::bigint as pr_number, + d ->> '$.user.login' as author_login, + d ->> '$.state' as state, + (d ->> '$.created_at')::timestamp as created_at, + (d ->> '$.merged_at')::timestamp as merged_at, + (d ->> '$.additions')::bigint as additions, + (d ->> '$.deletions')::bigint as deletions, + (d ->> '$.review_comments')::bigint as review_comments, + (d ->> '$.comments')::bigint as comment_count, + case + when (d ->> '$.merged_at') is not null then datediff( + 'hour', + (d ->> '$.created_at')::timestamp, + (d ->> '$.merged_at')::timestamp + ) + end as time_to_merge_hours, + d -> '$.labels' as labels, + _loaded_at +from source diff --git a/dbt/models/staging/github/stg_github__releases.sql b/dbt/models/staging/github/stg_github__releases.sql new file mode 100644 index 0000000..ceb118f --- /dev/null +++ b/dbt/models/staging/github/stg_github__releases.sql @@ -0,0 +1,13 @@ +with source as ( + select data::json as d, _repo, _loaded_at + from {{ source('github', 'releases') }} +) + +select + (d ->> '$.id')::bigint as release_id, + _repo as repository, + d ->> '$.tag_name' as tag_name, + d ->> '$.name' as name, + (d ->> '$.published_at')::timestamp as published_at, + _loaded_at +from source diff --git a/dbt/models/staging/github/stg_github__repositories.sql b/dbt/models/staging/github/stg_github__repositories.sql new file mode 100644 index 0000000..ecd7011 --- /dev/null +++ b/dbt/models/staging/github/stg_github__repositories.sql @@ -0,0 +1,21 @@ +with source as ( + select data::json as d, _loaded_at + from {{ source('github', 'repositories') }} +) + +select + (d ->> '$.id')::bigint as repository_id, + d ->> '$.full_name' as repository_name, + d ->> '$.name' as name, + d ->> '$.owner.login' as owner_login, + d ->> '$.description' as description, + (d ->> '$.stargazers_count')::bigint as stars, + (d ->> '$.forks_count')::bigint as forks, + (d ->> '$.open_issues_count')::bigint as open_issues, + d ->> '$.language' as language, + d ->> '$.license.spdx_id' as license_spdx, + d -> '$.topics' as topics, + (d ->> '$.created_at')::timestamp as created_at, + (d ->> '$.updated_at')::timestamp as updated_at, + _loaded_at +from source diff --git a/dbt/models/staging/pypi/_pypi__models.yml b/dbt/models/staging/pypi/_pypi__models.yml new file mode 100644 index 0000000..f7e907d --- /dev/null +++ b/dbt/models/staging/pypi/_pypi__models.yml @@ -0,0 +1,16 @@ +version: 2 + +models: + - name: stg_pypi__downloads + description: One cleaned row per package/category/day download count. + columns: + - name: package + data_tests: [not_null] + - 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] diff --git a/dbt/models/staging/pypi/_pypi__sources.yml b/dbt/models/staging/pypi/_pypi__sources.yml new file mode 100644 index 0000000..189caf4 --- /dev/null +++ b/dbt/models/staging/pypi/_pypi__sources.yml @@ -0,0 +1,13 @@ +version: 2 + +sources: + - name: pypi + description: Raw PyPI Stats responses; one row per package/day/category. + schema: raw + loaded_at_field: _loaded_at + freshness: + warn_after: {count: 12, period: hour} + error_after: {count: 48, period: hour} + tables: + - name: downloads + description: Daily download counts; `_package` carries the source package. diff --git a/dbt/models/staging/pypi/stg_pypi__downloads.sql b/dbt/models/staging/pypi/stg_pypi__downloads.sql new file mode 100644 index 0000000..979d20b --- /dev/null +++ b/dbt/models/staging/pypi/stg_pypi__downloads.sql @@ -0,0 +1,12 @@ +with source as ( + select data::json as d, _package, _loaded_at + from {{ source('pypi', 'downloads') }} +) + +select + _package as package, + d ->> '$.category' as category, + (d ->> '$.date')::date as download_date, + (d ->> '$.downloads')::bigint as download_count, + _loaded_at +from source diff --git a/dbt/package-lock.yml b/dbt/package-lock.yml new file mode 100644 index 0000000..b4c280e --- /dev/null +++ b/dbt/package-lock.yml @@ -0,0 +1,5 @@ +packages: + - name: dbt_utils + package: dbt-labs/dbt_utils + version: 1.3.3 +sha1_hash: a4da77dcded39caf20bd661f0098cbffd9735800 diff --git a/dbt/packages.yml b/dbt/packages.yml new file mode 100644 index 0000000..4b0e77e --- /dev/null +++ b/dbt/packages.yml @@ -0,0 +1,3 @@ +packages: + - package: dbt-labs/dbt_utils + version: [">=1.3.0", "<2.0.0"] diff --git a/dbt/profiles.yml b/dbt/profiles.yml new file mode 100644 index 0000000..8e16857 --- /dev/null +++ b/dbt/profiles.yml @@ -0,0 +1,12 @@ +repolytics: + target: dev + outputs: + dev: + type: duckdb + path: "{{ env_var('DUCKDB_PATH', '../data/warehouse/repolytics.duckdb') }}" + threads: 4 + ci: + # Ephemeral in-memory DB: enough for `dbt compile` / parsing, no data needed. + type: duckdb + path: ":memory:" + threads: 4 diff --git a/src/repolytics/loading/raw_loader.py b/src/repolytics/loading/raw_loader.py new file mode 100644 index 0000000..622ee35 --- /dev/null +++ b/src/repolytics/loading/raw_loader.py @@ -0,0 +1,54 @@ +"""Load landed Parquet into the DuckDB `raw` schema.""" + +from pathlib import Path + +import duckdb + +from repolytics.config import Settings, get_settings + +RAW_SCHEMA = "raw" + + +def load_raw(conn: duckdb.DuckDBPyConnection, raw_root: str | Path) -> list[str]: + """Load every `{source}/{table}` partition under `raw_root` into `raw.{table}`. + + Tables are (re)created with `CREATE OR REPLACE`, so the load is idempotent. + `union_by_name` tolerates partitions written with or without optional + metadata columns. Returns the sorted list of loaded table names. + """ + conn.execute(f"CREATE SCHEMA IF NOT EXISTS {RAW_SCHEMA}") + loaded: list[str] = [] + for table_dir in _table_dirs(Path(raw_root)): + table = table_dir.name + pattern = (table_dir / "**" / "*.parquet").as_posix() + conn.execute( + f'CREATE OR REPLACE TABLE {RAW_SCHEMA}."{table}" AS ' + "SELECT * FROM read_parquet(?, union_by_name = true)", + [pattern], + ) + loaded.append(table) + return loaded + + +def _table_dirs(root: Path) -> list[Path]: + """Return sorted `{source}/{table}` dirs under `root` that hold Parquet.""" + if not root.exists(): + return [] + return sorted( + table_dir + for source_dir in root.iterdir() + if source_dir.is_dir() + for table_dir in source_dir.iterdir() + if table_dir.is_dir() and next(table_dir.glob("**/*.parquet"), None) + ) + + +def load_all(settings: Settings | None = None) -> list[str]: + """Open the configured DuckDB file and load all raw partitions into it.""" + settings = settings or get_settings() + settings.duckdb_path.parent.mkdir(parents=True, exist_ok=True) + conn = duckdb.connect(str(settings.duckdb_path)) + try: + return load_raw(conn, settings.raw_data_path) + finally: + conn.close() diff --git a/tests/fixtures/.gitkeep b/tests/fixtures/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/tests/fixtures/github_responses/commits.json b/tests/fixtures/github_responses/commits.json new file mode 100644 index 0000000..3de7ca4 --- /dev/null +++ b/tests/fixtures/github_responses/commits.json @@ -0,0 +1,20 @@ +[ + { + "sha": "abc123", + "commit": { + "author": {"name": "Alice", "email": "alice@example.com", "date": "2024-01-02T10:00:00Z"}, + "message": "fix: handle empty response" + }, + "author": {"login": "alice", "id": 1}, + "stats": {"additions": 10, "deletions": 2, "total": 12} + }, + { + "sha": "def456", + "commit": { + "author": {"name": "Bob", "email": "bob@example.com", "date": "2024-01-03T12:00:00Z"}, + "message": "feat: add retry support" + }, + "author": {"login": "bob", "id": 2}, + "stats": {"additions": 30, "deletions": 5, "total": 35} + } +] diff --git a/tests/fixtures/github_responses/issues.json b/tests/fixtures/github_responses/issues.json new file mode 100644 index 0000000..907b6bc --- /dev/null +++ b/tests/fixtures/github_responses/issues.json @@ -0,0 +1,23 @@ +[ + { + "number": 100, + "title": "Crash on empty body", + "state": "closed", + "user": {"login": "carol", "id": 3}, + "created_at": "2024-01-01T00:00:00Z", + "closed_at": "2024-01-05T00:00:00Z", + "comments": 4, + "labels": [{"id": 1, "name": "bug", "color": "d73a4a"}] + }, + { + "number": 101, + "title": "Add feature (this is actually a PR)", + "state": "open", + "user": {"login": "dave", "id": 4}, + "created_at": "2024-01-02T00:00:00Z", + "closed_at": null, + "comments": 0, + "labels": [], + "pull_request": {"url": "https://api.github.com/repos/encode/httpx/pulls/101"} + } +] diff --git a/tests/fixtures/github_responses/pulls.json b/tests/fixtures/github_responses/pulls.json new file mode 100644 index 0000000..05f79e1 --- /dev/null +++ b/tests/fixtures/github_responses/pulls.json @@ -0,0 +1,30 @@ +[ + { + "number": 101, + "title": "Add feature", + "state": "closed", + "user": {"login": "dave", "id": 4}, + "created_at": "2024-01-02T00:00:00Z", + "updated_at": "2024-01-06T00:00:00Z", + "merged_at": "2024-01-06T00:00:00Z", + "additions": 120, + "deletions": 30, + "review_comments": 5, + "comments": 3, + "labels": [{"id": 2, "name": "enhancement", "color": "a2eeef"}] + }, + { + "number": 102, + "title": "Update docs", + "state": "open", + "user": {"login": "alice", "id": 1}, + "created_at": "2024-01-04T00:00:00Z", + "updated_at": "2024-01-04T00:00:00Z", + "merged_at": null, + "additions": 10, + "deletions": 1, + "review_comments": 0, + "comments": 0, + "labels": [] + } +] diff --git a/tests/fixtures/github_responses/releases.json b/tests/fixtures/github_responses/releases.json new file mode 100644 index 0000000..221d320 --- /dev/null +++ b/tests/fixtures/github_responses/releases.json @@ -0,0 +1,16 @@ +[ + { + "id": 1, + "tag_name": "0.27.0", + "name": "Version 0.27.0", + "published_at": "2024-01-01T00:00:00Z", + "assets": [] + }, + { + "id": 2, + "tag_name": "0.26.0", + "name": "Version 0.26.0", + "published_at": "2023-12-01T00:00:00Z", + "assets": [] + } +] diff --git a/tests/fixtures/github_responses/repository.json b/tests/fixtures/github_responses/repository.json new file mode 100644 index 0000000..0db663f --- /dev/null +++ b/tests/fixtures/github_responses/repository.json @@ -0,0 +1,15 @@ +{ + "id": 11730342, + "name": "httpx", + "full_name": "encode/httpx", + "owner": {"login": "encode", "id": 19315}, + "description": "A next generation HTTP client for Python.", + "stargazers_count": 13000, + "forks_count": 850, + "open_issues_count": 70, + "language": "Python", + "license": {"key": "bsd-3-clause", "spdx_id": "BSD-3-Clause"}, + "topics": ["python", "http", "asyncio"], + "created_at": "2013-07-28T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z" +} diff --git a/tests/fixtures/pypi_responses/overall.json b/tests/fixtures/pypi_responses/overall.json new file mode 100644 index 0000000..f97e647 --- /dev/null +++ b/tests/fixtures/pypi_responses/overall.json @@ -0,0 +1,8 @@ +{ + "data": [ + {"category": "without_mirrors", "date": "2024-01-01", "downloads": 12000}, + {"category": "without_mirrors", "date": "2024-01-02", "downloads": 13500} + ], + "package": "polars", + "type": "overall_downloads" +} diff --git a/tests/fixtures/pypi_responses/recent.json b/tests/fixtures/pypi_responses/recent.json new file mode 100644 index 0000000..86c9d96 --- /dev/null +++ b/tests/fixtures/pypi_responses/recent.json @@ -0,0 +1,5 @@ +{ + "data": {"last_day": 12345, "last_week": 80000, "last_month": 350000}, + "package": "polars", + "type": "recent_downloads" +} diff --git a/tests/integration/test_raw_loader.py b/tests/integration/test_raw_loader.py new file mode 100644 index 0000000..e27ed3b --- /dev/null +++ b/tests/integration/test_raw_loader.py @@ -0,0 +1,121 @@ +"""Integration tests for repolytics.loading.raw_loader.""" + +import json +from pathlib import Path + +import duckdb +import pytest + +from repolytics.ingestion.writer import partition_path, write_parquet +from repolytics.loading.raw_loader import load_raw + +FIXTURES = Path(__file__).parent.parent / "fixtures" +REPO = "encode/httpx" +PACKAGE = "polars" +DAY = "2024-01-02" + +# Source table -> (fixtures file, metadata). +_GITHUB_TABLES = { + "commits": "commits.json", + "issues": "issues.json", + "pull_requests": "pulls.json", + "releases": "releases.json", +} + + +def _read_json(rel: str) -> object: + return json.loads((FIXTURES / rel).read_text(encoding="utf-8")) + + +@pytest.fixture +def raw_root(tmp_data_dir: Path) -> Path: + """A populated `data/raw` tree built from the JSON fixtures via write_parquet.""" + root = tmp_data_dir / "raw" + + repo = _read_json("github_responses/repository.json") + write_parquet( + [repo], + partition_path(root, "github", "repositories", DAY, entity=REPO), + metadata={"_repo": REPO}, + ) + for table, fname in _GITHUB_TABLES.items(): + write_parquet( + _read_json(f"github_responses/{fname}"), + partition_path(root, "github", table, DAY, entity=REPO), + metadata={"_repo": REPO}, + ) + overall = _read_json("pypi_responses/overall.json") + write_parquet( + overall["data"], + partition_path(root, "pypi", "downloads", DAY, entity=PACKAGE), + metadata={"_package": PACKAGE}, + ) + return root + + +def test_load_raw_creates_expected_tables( + raw_root: Path, duckdb_conn: duckdb.DuckDBPyConnection +) -> None: + loaded = load_raw(duckdb_conn, raw_root) + + assert set(loaded) == { + "repositories", + "commits", + "issues", + "pull_requests", + "releases", + "downloads", + } + schemas = duckdb_conn.execute( + "SELECT schema_name FROM information_schema.schemata" + ).fetchall() + assert "raw" in {row[0] for row in schemas} + + +def test_load_raw_row_counts( + raw_root: Path, duckdb_conn: duckdb.DuckDBPyConnection +) -> None: + load_raw(duckdb_conn, raw_root) + + def count(table: str) -> int: + return duckdb_conn.execute(f"SELECT count(*) FROM raw.{table}").fetchone()[0] + + assert count("repositories") == 1 + assert count("commits") == 2 + assert count("issues") == 2 # includes the PR-shaped issue (staging filters it) + assert count("downloads") == 2 + + +def test_load_raw_preserves_data_and_metadata_columns( + raw_root: Path, duckdb_conn: duckdb.DuckDBPyConnection +) -> None: + load_raw(duckdb_conn, raw_root) + + columns = duckdb_conn.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = 'raw' AND table_name = 'commits'" + ).fetchall() + assert {"data", "_loaded_at", "_repo"} <= {row[0] for row in columns} + + repo = duckdb_conn.execute("SELECT DISTINCT _repo FROM raw.commits").fetchone()[0] + assert repo == REPO + package = duckdb_conn.execute( + "SELECT DISTINCT _package FROM raw.downloads" + ).fetchone()[0] + assert package == PACKAGE + + +def test_load_raw_is_idempotent( + raw_root: Path, duckdb_conn: duckdb.DuckDBPyConnection +) -> None: + load_raw(duckdb_conn, raw_root) + load_raw(duckdb_conn, raw_root) # second run must not duplicate rows + + count = duckdb_conn.execute("SELECT count(*) FROM raw.commits").fetchone()[0] + assert count == 2 + + +def test_load_raw_empty_root_returns_nothing( + tmp_data_dir: Path, duckdb_conn: duckdb.DuckDBPyConnection +) -> None: + assert load_raw(duckdb_conn, tmp_data_dir / "raw") == []