From cb5b334245d2fb4367ff1dcc6d313963ef449db2 Mon Sep 17 00:00:00 2001 From: Kenan Salim Date: Sun, 14 Jun 2026 00:50:04 +0300 Subject: [PATCH 1/3] data: fix #1318 type mismatch, enforce collab-silver contract, add dbt-coverage gate + CI Fixes the NO_COMMON_TYPE (Float64 vs Int64) build failure in class_collab_document_activity: visited_page_count is pinned to Nullable(Float64) on both M365 halves (OneDrive cast NULL; SharePoint cast through). Enables an ENFORCED dbt contract on class_collab_document_activity (every column declared with its data_type from the live ClickHouse catalog; on_schema_change: fail) so column/type drift fails the build, not production. First model on the path to full contract coverage. scripts/ci/dbt_coverage.py: walks the compiled manifest and flags metric-path models (bronze-promoted -> silver) lacking an enforced contract or not_null+unique on their key column, plus bronze source-freshness coverage. Validated locally with dbt-core 1.10 + dbt-clickhouse 1.9 (dbt parse). .github/workflows/data-contracts.yml: runs the gate in PR CI via dbt parse only (no warehouse/credentials), report-only, uploading the per-model gap list. Wording is neutral throughout (user-facing terms only). Closes #1318. Related: #1319, #1321, #1326. Signed-off-by: Kenan Salim --- .github/workflows/data-contracts.yml | 80 ++++++++++ scripts/ci/dbt_coverage.py | 139 ++++++++++++++++++ ...365__collab_document_activity_onedrive.sql | 5 +- ...5__collab_document_activity_sharepoint.sql | 5 +- src/ingestion/silver/collaboration/schema.yml | 58 +++++++- 5 files changed, 283 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/data-contracts.yml create mode 100644 scripts/ci/dbt_coverage.py diff --git a/.github/workflows/data-contracts.yml b/.github/workflows/data-contracts.yml new file mode 100644 index 000000000..371f8ab6e --- /dev/null +++ b/.github/workflows/data-contracts.yml @@ -0,0 +1,80 @@ +# Data-contract & test-completeness coverage — the QA gate that MEASURES, in CI, +# how much of the metric path is under an enforced dbt contract + key tests, and +# how many bronze sources have a freshness check. +# +# Runs on `dbt parse` only — NO database, NO credentials, NO seeded data needed +# (contract.enforced, tests, and source freshness all live in the parsed +# manifest). So every MR gets the numbers + the per-model gap list as a +# downloadable artifact and a job summary. +# +# Today it is REPORT-ONLY (informational): the baseline is 1/124 contracted, so +# blocking would red every PR. Dev burns the debt down (adds contracts); flip +# the marked step to `--check` per domain to make it blocking as coverage rises. +# Tracked as a QA-logged bug; QA owns this gate, Dev owns filling the contracts. + +name: Data Contracts + +on: + pull_request: + branches: [main] + paths: + - "src/ingestion/**" + - "scripts/ci/dbt_coverage.py" + - ".github/workflows/data-contracts.yml" + workflow_dispatch: + +permissions: + contents: read + +jobs: + dbt-coverage: + name: dbt-coverage + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + with: + persist-credentials: false + + - uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b # v5.3.0 + with: + python-version: "3.12" + + - name: Install dbt (core + clickhouse adapter) + run: pip install --quiet 'dbt-core~=1.10' 'dbt-clickhouse~=1.9' + + - name: dbt parse (manifest only — no warehouse connection) + working-directory: src/ingestion/dbt + env: + # parse does NOT connect; a placeholder satisfies the profile's env_var. + CLICKHOUSE_PASSWORD: ci-noconnect + DBT_PROFILES_DIR: ${{ github.workspace }}/src/ingestion/dbt + run: | + [ -f packages.yml ] && dbt deps || true + dbt parse --no-version-check + + - name: Data-contract & test coverage report + run: | + python scripts/ci/dbt_coverage.py src/ingestion/dbt/target/manifest.json \ + | tee /tmp/dbt-coverage.txt + { + echo '## dbt data-contract coverage' + echo '```' + grep -E 'metric-path models|enforced contract|key tests|bronze sources|freshness declared' /tmp/dbt-coverage.txt + echo '```' + echo '_Report-only. Dev fills contracts; flip the gate step to `--check` per domain to enforce._' + } >> "$GITHUB_STEP_SUMMARY" + + # ── Flip to blocking per domain as contracts land (QA ratchet) ── + # - name: Enforce (blocking) + # run: python scripts/ci/dbt_coverage.py src/ingestion/dbt/target/manifest.json --check + + - name: Upload coverage report + manifest (downloadable results) + if: always() + uses: actions/upload-artifact@b4b15b8c7c6ac21ea08fcf65892d2ee8f75cf882 # v4.4.3 + with: + name: dbt-coverage + path: | + /tmp/dbt-coverage.txt + src/ingestion/dbt/target/manifest.json + if-no-files-found: warn diff --git a/scripts/ci/dbt_coverage.py b/scripts/ci/dbt_coverage.py new file mode 100644 index 000000000..f085db8b2 --- /dev/null +++ b/scripts/ci/dbt_coverage.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""dbt_coverage.py — data-contract & test-completeness gate (the path to dbt 100%). + +Walks the compiled dbt manifest and asserts that every model in the path to an +user-facing metric (schema silver/gold, plus staging models that feed a +`silver:*` tag-union) carries: + + 1. an ENFORCED contract — config.contract.enforced == true, i.e. declared + column names + data_types. This is what turns a type drift like issue + #1318 (Float64 vs Int64) into a BUILD failure instead of a runtime + ClickHouse NO_COMMON_TYPE error; and + 2. key data tests — not_null + unique on its key. + +"100%" = zero metric-path models below this bar. + +Usage: + dbt_coverage.py [MANIFEST] # report coverage, exit 0 + dbt_coverage.py [MANIFEST] --check # exit 1 if any metric-path model has a gap + +Needs a compiled manifest (default src/ingestion/dbt/target/manifest.json) — +run `dbt parse` (or any build/compile) in the dbt project first. +""" +import json +import pathlib +import sys + +DEFAULT_MANIFEST = pathlib.Path("src/ingestion/dbt/target/manifest.json") + + +def is_metric_path(node: dict) -> bool: + """A dbt model on the path to an user-facing metric: silver/gold, the + silver:* tag-union feeders, AND bronze-promotion views (the first typed + boundary). NOTE: gold serving tables here are ClickHouse views built by + analytics-api migrations, not dbt — their contract lives in the API-contract + track (openapi/schemathesis), not this gate.""" + schema = (node.get("config") or {}).get("schema") or node.get("schema") or "" + name = node.get("name") or "" + tags = node.get("tags") or [] + return ( + schema in ("silver", "gold", "bronze") + or name.endswith("__bronze_promoted") + or any(t == "silver" or t.startswith("silver:") for t in tags) + ) + + +def main() -> None: + check = "--check" in sys.argv + positional = [a for a in sys.argv[1:] if not a.startswith("--")] + manifest_path = pathlib.Path(positional[0]) if positional else DEFAULT_MANIFEST + + if not manifest_path.exists(): + print(f"✗ manifest not found: {manifest_path}") + print(" compile it first: (cd src/ingestion/dbt && dbt parse)") + sys.exit(2) + + manifest = json.loads(manifest_path.read_text()) + nodes = manifest.get("nodes", {}) + + # --- Bronze sources: every Airbyte source should declare freshness --- + sources = manifest.get("sources", {}) + if sources: + fresh = sum(1 for s in sources.values() if s.get("freshness") and s.get("loaded_at_field")) + print(f"bronze sources: {len(sources)} — freshness declared: {fresh}/{len(sources)} " + f"({100 * fresh // len(sources)}%)") + for uid, s in sorted(sources.items(), key=lambda kv: kv[1].get("name", "")): + if not (s.get("freshness") and s.get("loaded_at_field")): + print(f" ✗ source {s.get('source_name')}.{s.get('name')}: no freshness check") + + models = { + uid: n + for uid, n in nodes.items() + if n.get("resource_type") == "model" and is_metric_path(n) + } + + # (model uid, column) -> set of test kinds, so we can check the ACTUAL key + # column is tested, not just that the model has *some* not_null + *some* + # unique somewhere (which a model can satisfy on unrelated columns). + col_tests: dict[tuple, set] = {} + for n in nodes.values(): + if n.get("resource_type") != "test": + continue + kind = (n.get("test_metadata") or {}).get("name") or "" + col = n.get("column_name") or ( + (n.get("test_metadata") or {}).get("kwargs") or {} + ).get("column_name") + if not col: + continue + for dep in (n.get("depends_on") or {}).get("nodes", []): + col_tests.setdefault((dep, col), set()).add(kind) + + def key_columns(node: dict) -> list: + key = (node.get("config") or {}).get("unique_key") + if isinstance(key, str): + return [key] + if isinstance(key, list) and key: + return key + return ["unique_key"] + + gaps, contract_ok, tests_ok = [], 0, 0 + for uid, n in sorted(models.items(), key=lambda kv: kv[1].get("name", "")): + # contract.enforced lives at the node top level AND under config in + # current dbt; read both so the check is correct across versions. + enforced = bool( + (n.get("contract") or {}).get("enforced") + or ((n.get("config") or {}).get("contract") or {}).get("enforced") + ) + keys = key_columns(n) + has_keys = all( + "not_null" in col_tests.get((uid, col), set()) + and "unique" in col_tests.get((uid, col), set()) + for col in keys + ) + contract_ok += enforced + tests_ok += has_keys + problems = [] + if not enforced: + problems.append("no enforced contract") + if not has_keys: + problems.append(f"missing not_null+unique on key column(s) {keys}") + if problems: + gaps.append((n.get("name"), problems)) + + total = len(models) + print(f"metric-path models: {total}") + if total: + print(f" enforced contract: {contract_ok}/{total} ({100 * contract_ok // total}%)") + print(f" key tests : {tests_ok}/{total} ({100 * tests_ok // total}%)") + for name, problems in gaps: + print(f" ✗ {name}: {', '.join(problems)}") + + if check and gaps: + print(f"\n✗ data-contract gate FAILED: {len(gaps)} metric-path model(s) below the contract.") + sys.exit(1) + if check: + print("✓ data-contract gate passed — 100% of metric-path models contracted + key-tested") + + +if __name__ == "__main__": + main() diff --git a/src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_onedrive.sql b/src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_onedrive.sql index e58e79050..c3c4572bf 100644 --- a/src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_onedrive.sql +++ b/src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_onedrive.sql @@ -28,7 +28,10 @@ SELECT syncedFileCount AS synced_count, sharedInternallyFileCount AS shared_internally_count, sharedExternallyFileCount AS shared_externally_count, - CAST(NULL AS Nullable(Int64)) AS visited_page_count, + -- OneDrive has no page-visit metric. Pinned to Nullable(Float64) to match + -- the SharePoint half: the silver tag-union (class_collab_document_activity) + -- fails with NO_COMMON_TYPE if the two halves disagree (Int64 vs Float64). + CAST(NULL AS Nullable(Float64)) AS visited_page_count, reportPeriod AS report_period, now() AS collected_at, 'insight_m365' AS data_source, diff --git a/src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_sharepoint.sql b/src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_sharepoint.sql index ce9b7d059..1b5a614ae 100644 --- a/src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_sharepoint.sql +++ b/src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_sharepoint.sql @@ -28,7 +28,10 @@ SELECT syncedFileCount AS synced_count, sharedInternallyFileCount AS shared_internally_count, sharedExternallyFileCount AS shared_externally_count, - visitedPageCount AS visited_page_count, + -- Pinned to Nullable(Float64) so it matches the OneDrive half (which has no + -- page-visit metric): the silver tag-union fails with NO_COMMON_TYPE if the + -- bronze Float64 here meets a bare Int64 there. + CAST(visitedPageCount AS Nullable(Float64)) AS visited_page_count, reportPeriod AS report_period, now() AS collected_at, 'insight_m365' AS data_source, diff --git a/src/ingestion/silver/collaboration/schema.yml b/src/ingestion/silver/collaboration/schema.yml index a0e1ddf69..61bce75a8 100644 --- a/src/ingestion/silver/collaboration/schema.yml +++ b/src/ingestion/silver/collaboration/schema.yml @@ -253,36 +253,90 @@ models: - not_null - name: class_collab_document_activity - description: "Unified daily document activity per user (M365 OneDrive + SharePoint)" + description: > + Unified daily document activity per user (M365 OneDrive + SharePoint). + + ENFORCED CONTRACT (data_type per column): dbt verifies the built relation + against these declared types at build time. This is what makes the + Float64-vs-Int64 class (issue #1318) a build failure instead of a runtime + ClickHouse NO_COMMON_TYPE error. Types are the live ClickHouse catalog + types; keep them in sync if a staging column's type legitimately changes + (and update both M365 halves together so the tag-union stays type-stable). + config: + contract: + enforced: true + # Contract-enforced incremental models must not silently ignore schema + # drift; fail the run so a column/type change is caught, not absorbed. + on_schema_change: fail columns: - name: tenant_id + data_type: String description: "Tenant isolation field" tests: - not_null - name: insight_source_id + data_type: String description: "Source instance identifier" tests: - not_null - name: unique_key - description: "Composite deduplication key" + data_type: FixedString(16) + description: "Composite dedup key (MD5 → FixedString(16))" tests: - not_null - unique + - name: user_id + data_type: String + description: "userPrincipalName" + - name: user_name + data_type: String + description: "userPrincipalName" + - name: email + data_type: String + description: "userPrincipalName" - name: person_key + data_type: String description: "lower(email) with lower(user_name) fallback — cross-source identity join key" tests: - not_null - name: date + data_type: Date description: "Activity date (UTC)" tests: - not_null - name: product + data_type: String description: "onedrive | sharepoint" tests: - not_null - accepted_values: values: ['onedrive', 'sharepoint'] + - name: viewed_or_edited_count + data_type: Nullable(Float64) + description: "Files viewed or edited" + - name: synced_count + data_type: Nullable(Float64) + description: "Files synced" + - name: shared_internally_count + data_type: Nullable(Float64) + description: "Files shared internally" + - name: shared_externally_count + data_type: Nullable(Float64) + description: "Files shared externally" + - name: visited_page_count + data_type: Nullable(Float64) + description: "SharePoint page visits (NULL for OneDrive). Pinned Float64 on both halves — see #1318." + - name: report_period + data_type: Nullable(String) + description: "M365 report period (days)" + - name: collected_at + data_type: DateTime + description: "Ingestion timestamp" - name: data_source + data_type: String description: "Source discriminator: insight_m365" tests: - not_null + - name: _version + data_type: Int64 + description: "ReplacingMergeTree version (latest wins)" From 6fab20e4660af9f52e8df852a545c6c3ac63e6ed Mon Sep 17 00:00:00 2001 From: Kenan Salim Date: Mon, 22 Jun 2026 18:23:12 +0300 Subject: [PATCH 2/3] data: address review on dbt-coverage gate - is_metric_path: drop blanket bronze-schema scope; keep bronze-promotion views (__bronze_promoted) + silver/gold + silver:* feeders so the denominator matches the gate's stated scope. - sources freshness: relabel 'bronze sources' -> 'sources' (it counts all declared sources) and fix unused loop var (ruff B007). - CI: fail fast on 'dbt deps' instead of '|| true'; guard summary grep with '|| true' so harmless output drift can't red the step. Signed-off-by: Kenan Salim --- .github/workflows/data-contracts.yml | 6 ++++-- scripts/ci/dbt_coverage.py | 8 ++++---- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/.github/workflows/data-contracts.yml b/.github/workflows/data-contracts.yml index 371f8ab6e..c8559a788 100644 --- a/.github/workflows/data-contracts.yml +++ b/.github/workflows/data-contracts.yml @@ -50,7 +50,9 @@ jobs: CLICKHOUSE_PASSWORD: ci-noconnect DBT_PROFILES_DIR: ${{ github.workspace }}/src/ingestion/dbt run: | - [ -f packages.yml ] && dbt deps || true + if [ -f packages.yml ]; then + dbt deps + fi dbt parse --no-version-check - name: Data-contract & test coverage report @@ -60,7 +62,7 @@ jobs: { echo '## dbt data-contract coverage' echo '```' - grep -E 'metric-path models|enforced contract|key tests|bronze sources|freshness declared' /tmp/dbt-coverage.txt + grep -E 'metric-path models|enforced contract|key tests|^sources:|freshness declared' /tmp/dbt-coverage.txt || true echo '```' echo '_Report-only. Dev fills contracts; flip the gate step to `--check` per domain to enforce._' } >> "$GITHUB_STEP_SUMMARY" diff --git a/scripts/ci/dbt_coverage.py b/scripts/ci/dbt_coverage.py index f085db8b2..827c67254 100644 --- a/scripts/ci/dbt_coverage.py +++ b/scripts/ci/dbt_coverage.py @@ -37,7 +37,7 @@ def is_metric_path(node: dict) -> bool: name = node.get("name") or "" tags = node.get("tags") or [] return ( - schema in ("silver", "gold", "bronze") + schema in ("silver", "gold") or name.endswith("__bronze_promoted") or any(t == "silver" or t.startswith("silver:") for t in tags) ) @@ -56,13 +56,13 @@ def main() -> None: manifest = json.loads(manifest_path.read_text()) nodes = manifest.get("nodes", {}) - # --- Bronze sources: every Airbyte source should declare freshness --- + # --- Sources: every declared source should carry a freshness check --- sources = manifest.get("sources", {}) if sources: fresh = sum(1 for s in sources.values() if s.get("freshness") and s.get("loaded_at_field")) - print(f"bronze sources: {len(sources)} — freshness declared: {fresh}/{len(sources)} " + print(f"sources: {len(sources)} — freshness declared: {fresh}/{len(sources)} " f"({100 * fresh // len(sources)}%)") - for uid, s in sorted(sources.items(), key=lambda kv: kv[1].get("name", "")): + for s in sorted(sources.values(), key=lambda s: s.get("name", "")): if not (s.get("freshness") and s.get("loaded_at_field")): print(f" ✗ source {s.get('source_name')}.{s.get('name')}: no freshness check") From 4b490a49b07f77c5694a9bd74590c733807c4b94 Mon Sep 17 00:00:00 2001 From: SharedQA <122366558+SharedQA@users.noreply.github.com> Date: Tue, 23 Jun 2026 11:26:00 +0300 Subject: [PATCH 3/3] ci(data): move dbt-coverage gate to the consolidated data-quality.yml (#1335) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop data-contracts.yml + dbt_coverage.py from this PR — they now live in the single data-quality.yml on #1335 (one data-gate workflow instead of three). This PR keeps only its substance: the m365 collab_document type-mismatch fix (#1318) + the silver collaboration contract. Signed-off-by: SharedQA <122366558+SharedQA@users.noreply.github.com> --- .github/workflows/data-contracts.yml | 82 ---------------- scripts/ci/dbt_coverage.py | 139 --------------------------- 2 files changed, 221 deletions(-) delete mode 100644 .github/workflows/data-contracts.yml delete mode 100644 scripts/ci/dbt_coverage.py diff --git a/.github/workflows/data-contracts.yml b/.github/workflows/data-contracts.yml deleted file mode 100644 index c8559a788..000000000 --- a/.github/workflows/data-contracts.yml +++ /dev/null @@ -1,82 +0,0 @@ -# Data-contract & test-completeness coverage — the QA gate that MEASURES, in CI, -# how much of the metric path is under an enforced dbt contract + key tests, and -# how many bronze sources have a freshness check. -# -# Runs on `dbt parse` only — NO database, NO credentials, NO seeded data needed -# (contract.enforced, tests, and source freshness all live in the parsed -# manifest). So every MR gets the numbers + the per-model gap list as a -# downloadable artifact and a job summary. -# -# Today it is REPORT-ONLY (informational): the baseline is 1/124 contracted, so -# blocking would red every PR. Dev burns the debt down (adds contracts); flip -# the marked step to `--check` per domain to make it blocking as coverage rises. -# Tracked as a QA-logged bug; QA owns this gate, Dev owns filling the contracts. - -name: Data Contracts - -on: - pull_request: - branches: [main] - paths: - - "src/ingestion/**" - - "scripts/ci/dbt_coverage.py" - - ".github/workflows/data-contracts.yml" - workflow_dispatch: - -permissions: - contents: read - -jobs: - dbt-coverage: - name: dbt-coverage - runs-on: ubuntu-latest - timeout-minutes: 15 - steps: - - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 - with: - persist-credentials: false - - - uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b # v5.3.0 - with: - python-version: "3.12" - - - name: Install dbt (core + clickhouse adapter) - run: pip install --quiet 'dbt-core~=1.10' 'dbt-clickhouse~=1.9' - - - name: dbt parse (manifest only — no warehouse connection) - working-directory: src/ingestion/dbt - env: - # parse does NOT connect; a placeholder satisfies the profile's env_var. - CLICKHOUSE_PASSWORD: ci-noconnect - DBT_PROFILES_DIR: ${{ github.workspace }}/src/ingestion/dbt - run: | - if [ -f packages.yml ]; then - dbt deps - fi - dbt parse --no-version-check - - - name: Data-contract & test coverage report - run: | - python scripts/ci/dbt_coverage.py src/ingestion/dbt/target/manifest.json \ - | tee /tmp/dbt-coverage.txt - { - echo '## dbt data-contract coverage' - echo '```' - grep -E 'metric-path models|enforced contract|key tests|^sources:|freshness declared' /tmp/dbt-coverage.txt || true - echo '```' - echo '_Report-only. Dev fills contracts; flip the gate step to `--check` per domain to enforce._' - } >> "$GITHUB_STEP_SUMMARY" - - # ── Flip to blocking per domain as contracts land (QA ratchet) ── - # - name: Enforce (blocking) - # run: python scripts/ci/dbt_coverage.py src/ingestion/dbt/target/manifest.json --check - - - name: Upload coverage report + manifest (downloadable results) - if: always() - uses: actions/upload-artifact@b4b15b8c7c6ac21ea08fcf65892d2ee8f75cf882 # v4.4.3 - with: - name: dbt-coverage - path: | - /tmp/dbt-coverage.txt - src/ingestion/dbt/target/manifest.json - if-no-files-found: warn diff --git a/scripts/ci/dbt_coverage.py b/scripts/ci/dbt_coverage.py deleted file mode 100644 index 827c67254..000000000 --- a/scripts/ci/dbt_coverage.py +++ /dev/null @@ -1,139 +0,0 @@ -#!/usr/bin/env python3 -"""dbt_coverage.py — data-contract & test-completeness gate (the path to dbt 100%). - -Walks the compiled dbt manifest and asserts that every model in the path to an -user-facing metric (schema silver/gold, plus staging models that feed a -`silver:*` tag-union) carries: - - 1. an ENFORCED contract — config.contract.enforced == true, i.e. declared - column names + data_types. This is what turns a type drift like issue - #1318 (Float64 vs Int64) into a BUILD failure instead of a runtime - ClickHouse NO_COMMON_TYPE error; and - 2. key data tests — not_null + unique on its key. - -"100%" = zero metric-path models below this bar. - -Usage: - dbt_coverage.py [MANIFEST] # report coverage, exit 0 - dbt_coverage.py [MANIFEST] --check # exit 1 if any metric-path model has a gap - -Needs a compiled manifest (default src/ingestion/dbt/target/manifest.json) — -run `dbt parse` (or any build/compile) in the dbt project first. -""" -import json -import pathlib -import sys - -DEFAULT_MANIFEST = pathlib.Path("src/ingestion/dbt/target/manifest.json") - - -def is_metric_path(node: dict) -> bool: - """A dbt model on the path to an user-facing metric: silver/gold, the - silver:* tag-union feeders, AND bronze-promotion views (the first typed - boundary). NOTE: gold serving tables here are ClickHouse views built by - analytics-api migrations, not dbt — their contract lives in the API-contract - track (openapi/schemathesis), not this gate.""" - schema = (node.get("config") or {}).get("schema") or node.get("schema") or "" - name = node.get("name") or "" - tags = node.get("tags") or [] - return ( - schema in ("silver", "gold") - or name.endswith("__bronze_promoted") - or any(t == "silver" or t.startswith("silver:") for t in tags) - ) - - -def main() -> None: - check = "--check" in sys.argv - positional = [a for a in sys.argv[1:] if not a.startswith("--")] - manifest_path = pathlib.Path(positional[0]) if positional else DEFAULT_MANIFEST - - if not manifest_path.exists(): - print(f"✗ manifest not found: {manifest_path}") - print(" compile it first: (cd src/ingestion/dbt && dbt parse)") - sys.exit(2) - - manifest = json.loads(manifest_path.read_text()) - nodes = manifest.get("nodes", {}) - - # --- Sources: every declared source should carry a freshness check --- - sources = manifest.get("sources", {}) - if sources: - fresh = sum(1 for s in sources.values() if s.get("freshness") and s.get("loaded_at_field")) - print(f"sources: {len(sources)} — freshness declared: {fresh}/{len(sources)} " - f"({100 * fresh // len(sources)}%)") - for s in sorted(sources.values(), key=lambda s: s.get("name", "")): - if not (s.get("freshness") and s.get("loaded_at_field")): - print(f" ✗ source {s.get('source_name')}.{s.get('name')}: no freshness check") - - models = { - uid: n - for uid, n in nodes.items() - if n.get("resource_type") == "model" and is_metric_path(n) - } - - # (model uid, column) -> set of test kinds, so we can check the ACTUAL key - # column is tested, not just that the model has *some* not_null + *some* - # unique somewhere (which a model can satisfy on unrelated columns). - col_tests: dict[tuple, set] = {} - for n in nodes.values(): - if n.get("resource_type") != "test": - continue - kind = (n.get("test_metadata") or {}).get("name") or "" - col = n.get("column_name") or ( - (n.get("test_metadata") or {}).get("kwargs") or {} - ).get("column_name") - if not col: - continue - for dep in (n.get("depends_on") or {}).get("nodes", []): - col_tests.setdefault((dep, col), set()).add(kind) - - def key_columns(node: dict) -> list: - key = (node.get("config") or {}).get("unique_key") - if isinstance(key, str): - return [key] - if isinstance(key, list) and key: - return key - return ["unique_key"] - - gaps, contract_ok, tests_ok = [], 0, 0 - for uid, n in sorted(models.items(), key=lambda kv: kv[1].get("name", "")): - # contract.enforced lives at the node top level AND under config in - # current dbt; read both so the check is correct across versions. - enforced = bool( - (n.get("contract") or {}).get("enforced") - or ((n.get("config") or {}).get("contract") or {}).get("enforced") - ) - keys = key_columns(n) - has_keys = all( - "not_null" in col_tests.get((uid, col), set()) - and "unique" in col_tests.get((uid, col), set()) - for col in keys - ) - contract_ok += enforced - tests_ok += has_keys - problems = [] - if not enforced: - problems.append("no enforced contract") - if not has_keys: - problems.append(f"missing not_null+unique on key column(s) {keys}") - if problems: - gaps.append((n.get("name"), problems)) - - total = len(models) - print(f"metric-path models: {total}") - if total: - print(f" enforced contract: {contract_ok}/{total} ({100 * contract_ok // total}%)") - print(f" key tests : {tests_ok}/{total} ({100 * tests_ok // total}%)") - for name, problems in gaps: - print(f" ✗ {name}: {', '.join(problems)}") - - if check and gaps: - print(f"\n✗ data-contract gate FAILED: {len(gaps)} metric-path model(s) below the contract.") - sys.exit(1) - if check: - print("✓ data-contract gate passed — 100% of metric-path models contracted + key-tested") - - -if __name__ == "__main__": - main()