Implementing Data Contracts in a dbt Monorepo

version: 1
model: fct_orders
owner: data-platform-team
consumers:
  - analytics-dashboards
  - ml-feature-pipeline
  - finance-reporting
guaranteed_columns:
  - name: order_id
    type: bigint
    nullable: false
    description: "Immutable surrogate key — never renamed or retyped"
  - name: customer_id
    type: bigint
    nullable: false
  - name: order_total_usd
    type: numeric(10,2)
    nullable: false
  - name: order_status
    type: varchar
    nullable: false
    allowed_values: [pending, confirmed, shipped, delivered, cancelled]
  - name: created_at
    type: timestamp with time zone
    nullable: false
freshness_sla_minutes: 60
breaking_change_policy: require_major_version_bump

models:
  - name: fct_orders
    config:
      contract:
        enforced: true   # dbt fails the run if column types don't match
    columns:
      - name: order_id
        data_type: bigint
        constraints:
          - type: not_null
          - type: primary_key
      - name: order_total_usd
        data_type: numeric
        constraints:
          - type: not_null

import yaml, sys, os
from pathlib import Path
import snowflake.connector

def load_contract(contract_path: str) -> dict:
    with open(contract_path) as f:
        return yaml.safe_load(f)

def get_warehouse_columns(conn, model: str) -> dict[str, dict]:
    """Query information_schema for the model's current column definitions."""
    cursor = conn.cursor()
    cursor.execute(f"""
        SELECT column_name, data_type, is_nullable
        FROM information_schema.columns
        WHERE table_name = UPPER('{model}')
          AND table_schema = UPPER(%(schema)s)
    """, {"schema": "analytics"})
    return {
        row[0].lower(): {"type": row[1].lower(), "nullable": row[2] == "YES"}
        for row in cursor.fetchall()
    }

def validate_contract(contract: dict, warehouse_cols: dict) -> list[str]:
    violations = []
    for col in contract["guaranteed_columns"]:
        name = col["name"]
        if name not in warehouse_cols:
            violations.append(f"MISSING: column '{name}' was dropped or renamed")
            continue
        actual = warehouse_cols[name]
        if col["type"].split("(")[0] not in actual["type"]:
            violations.append(
                f"TYPE CHANGE: '{name}' expected {col['type']}, got {actual['type']}"
            )
        if not col["nullable"] and actual["nullable"]:
            violations.append(
                f"NULLABILITY: '{name}' was non-nullable, now allows NULLs"
            )
    return violations

def main():
    conn = snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_CI_USER"],
        password=os.environ["SNOWFLAKE_CI_PASSWORD"],
    )
    contracts_dir = Path("contracts")
    all_violations = []

    for contract_file in contracts_dir.glob("*.yml"):
        contract = load_contract(contract_file)
        warehouse_cols = get_warehouse_columns(conn, contract["model"])
        violations = validate_contract(contract, warehouse_cols)
        if violations:
            print(f"\nContract violations in {contract_file.name}:")
            for v in violations:
                print(f"  ✗ {v}")
            all_violations.extend(violations)

    if all_violations:
        print(f"\n{len(all_violations)} contract violation(s) found. Merge blocked.")
        sys.exit(1)
    else:
        print("All contracts satisfied. ✓")

if __name__ == "__main__":
    main()

name: Data Contract Validation

on:
  pull_request:
    paths:
      - 'models/marts/**'
      - 'models/staging/**'
      - 'contracts/**'

jobs:
  validate-contracts:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }

      - name: Install dependencies
        run: pip install pyyaml snowflake-connector-python dbt-snowflake

      - name: Run dbt compile (CI slim run)
        run: dbt compile --select state:modified+ --defer --state ./prod-manifest
        env:
          DBT_PROFILES_DIR: .

      - name: Validate data contracts
        run: python scripts/validate_contracts.py
        env:
          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_CI_USER: ${{ secrets.SNOWFLAKE_CI_USER }}
          SNOWFLAKE_CI_PASSWORD: ${{ secrets.SNOWFLAKE_CI_PASSWORD }}

version: 2
model: fct_orders
migration_notes: "order_total renamed to order_total_usd in v2. Consumers have 30 days to migrate."
deprecated_columns:
  - name: order_total
    removed_in_version: 3
    replacement: order_total_usd
guaranteed_columns:
  - name: order_total_usd   # new name
    type: numeric(10,2)
    nullable: false
  - name: order_total       # keep old name as alias during migration window
    type: numeric(10,2)
    nullable: true
    deprecated: true