# A mental model: every quality check belongs to a dimension
quality_dimensions = {
    "completeness": ["not_null", "row_count_within_range", "no_missing_partitions"],
    "accuracy":     ["matches_reference_data", "reconciliation_with_source"],
    "consistency":  ["cross_system_totals_match", "referential_integrity"],
    "timeliness":   ["freshness_within_sla", "no_late_arriving_data"],
    "uniqueness":   ["unique_key", "no_duplicate_business_keys"],
    "validity":     ["accepted_values", "regex_match", "type_check", "range_check"],
}

# models/marts/schema.yml
version: 2

models:
  - name: fct_orders
    description: "One row per completed order"
    tests:
      # Model-level test: row count should not collapse
      - dbt_utils.equal_rowcount:
          compare_model: ref('stg_orders')
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: order_total
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1000000
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'shipped', 'delivered', 'cancelled']
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id

-- tests/assert_revenue_reconciles.sql
-- Fails if warehouse revenue diverges from the source ledger by >$1
WITH warehouse AS (
    SELECT date_trunc('day', ordered_at) AS day, SUM(order_total) AS rev
    FROM {{ ref('fct_orders') }}
    GROUP BY 1
),
ledger AS (
    SELECT day, SUM(amount) AS rev
    FROM {{ source('finance', 'daily_ledger') }}
    GROUP BY 1
)
SELECT w.day, w.rev AS warehouse_rev, l.rev AS ledger_rev
FROM warehouse w
JOIN ledger l USING (day)
WHERE ABS(w.rev - l.rev) > 1.00

import great_expectations as gx

context = gx.get_context()
batch = context.sources.pandas_default.read_csv("incoming_payments.csv")

# Completeness
batch.expect_column_values_to_not_be_null("payment_id")
batch.expect_table_row_count_to_be_between(min_value=100, max_value=1_000_000)

# Uniqueness
batch.expect_column_values_to_be_unique("payment_id")

# Validity
batch.expect_column_values_to_match_regex(
    "email", r"^[^@\s]+@[^@\s]+\.[^@\s]+$"
)
batch.expect_column_values_to_be_in_set(
    "currency", ["USD", "EUR", "GBP", "NZD", "AUD"]
)

# Accuracy / consistency between columns
batch.expect_column_pair_values_a_to_be_greater_than_b(
    "paid_at", "created_at", or_equal=True
)

# Distribution (statistical accuracy)
batch.expect_column_mean_to_be_between("amount", min_value=10, max_value=500)

results = batch.validate()
if not results.success:
    raise ValueError(f"Validation failed: {results['statistics']}")

# checks/orders.yml
checks for fct_orders:
  - row_count > 0
  - missing_count(customer_id) = 0
  - duplicate_count(order_id) = 0
  - invalid_percent(status) = 0:
      valid values: [pending, shipped, delivered, cancelled]
  - freshness(ordered_at) < 1h
  # Anomaly detection on a metric over time
  - anomaly score for row_count < default
  # Cross-check against another dataset
  - values in (customer_id) must exist in dim_customers (customer_id)

-- Flag days where row count deviates >3 standard deviations from the
-- trailing 30-day mean (a classic z-score check)
WITH daily AS (
    SELECT
        date_trunc('day', ordered_at) AS day,
        COUNT(*) AS row_count
    FROM fct_orders
    GROUP BY 1
),
stats AS (
    SELECT
        day,
        row_count,
        AVG(row_count)    OVER w AS mean_count,
        STDDEV(row_count) OVER w AS std_count
    FROM daily
    WINDOW w AS (ORDER BY day ROWS BETWEEN 30 PRECEDING AND 1 PRECEDING)
)
SELECT day, row_count, mean_count, std_count
FROM stats
WHERE ABS(row_count - mean_count) > 3 * std_count

# What to monitor for anomalies, by metric type
anomaly_targets = {
    "volume":       "row count per partition vs trailing baseline",
    "freshness":    "max(loaded_at) vs expected arrival time",
    "distribution": "mean/median/null-rate of key columns over time",
    "schema":       "column set and types vs last known good",
    "cardinality":  "distinct counts (e.g. active users) vs baseline",
}

Extract → Load to STAGING → AUDIT (run quality checks) → PUBLISH (swap into PROD)
                                  │
                                  └── fail → halt, alert, do not publish

# Write-Audit-Publish gate in an orchestrator (see ./orchestration)
def audit_then_publish(table: str):
    """Promote staged data only if it passes quality checks."""
    staging_table = f"staging.{table}"
    prod_table = f"analytics.{table}"

    failures = run_quality_checks(staging_table)  # returns list of failed checks
    blocking = [f for f in failures if f.severity == "error"]

    if blocking:
        send_alert(
            channel="#data-incidents",
            message=f"BLOCKED publish of {table}: {[f.name for f in blocking]}",
        )
        raise QualityGateError(f"{len(blocking)} blocking checks failed")

    # All blocking checks passed - atomic swap into production
    atomic_swap(staging_table, prod_table)

# data_sla.yml - a publishable contract for a critical dataset
dataset: analytics.fct_orders
owner: data-engineering
consumers: [finance, growth, exec-dashboard]

slos:
  freshness:
    sli: "time between event and availability in warehouse"
    objective: "p95 < 60 minutes"
    measurement_window: 30d
  completeness:
    sli: "percent of expected daily partitions present"
    objective: ">= 99.9%"
  accuracy:
    sli: "revenue reconciliation vs finance ledger"
    objective: "diff < 0.1% on 100% of days"

on_breach:
  notify: ["#data-incidents", "pagerduty:data-oncall"]
  status_page: true

# An actionable alert payload
alert = {
    "severity":       "error",
    "dataset":        "analytics.fct_orders",
    "check":          "freshness_within_sla",
    "detail":         "Latest partition is 3h old; SLO is 1h",
    "blast_radius":   ["exec_dashboard", "finance.daily_revenue", "growth.cohorts"],
    "owner":          "data-engineering",
    "runbook":        "https://wiki/runbooks/orders-freshness",
    "since":          "2026-05-30T07:00:00Z",
}