06Standards & CI/CD

from pyspark.sql import DataFrame, functions as F

# Business rules 1, 2, 5 and 6 as a plain dict — the pipeline and the
# test both import THIS, so they can never disagree about the rules.
# Rule 3 (valid countries DE/AT/CH/NL) is deliberately absent: an order
# line carries no country, so that rule is enforced in the customer
# transform. An expectation naming a column the table does not have
# fails the pipeline at analysis, not at runtime.
EXPECTATIONS = {
    "amount_positive":   "amount_eur > 0 OR status = 'CANCELLED'",
    "known_currency":    "currency IN ('EUR', 'CHF')",
    "fx_rate_resolved":  "fx_rate IS NOT NULL",
    "order_not_ancient": "order_ts >= TIMESTAMP '2020-01-01'",
    "status_domain":     "status IN ('OPEN', 'SHIPPED', 'CANCELLED')",
}

def to_silver_sales_order(lines: DataFrame, rates: DataFrame) -> DataFrame:
    """lines: bronze.sap_vbap shape. rates: silver.fx_rate (currency, rate_date, fx_rate)."""
    return (
        lines.withColumn("rate_date", F.to_date("order_ts"))
        .join(rates, on=["currency", "rate_date"], how="left")
        # EUR is the base currency and is absent from the ECB file, so it is 1
        # by definition. Any OTHER currency with no rate stays NULL on purpose:
        # fx_rate_resolved then drops the row and the event log counts it. The
        # literal is cast, because a Python float would promote the whole
        # DECIMAL chain to DOUBLE.
        .withColumn(
            "fx_rate",
            F.when(F.col("currency") == "EUR", F.lit(1).cast("decimal(18,6)"))
             .otherwise(F.col("fx_rate")),
        )
        .withColumn(
            "amount_eur",
            F.round(F.col("quantity") * F.col("unit_price") * F.col("fx_rate"), 2),
        )
        .drop("rate_date")
    )

from pyspark import pipelines as dp
from dwh.pipelines.transforms.silver_sales_order import EXPECTATIONS, to_silver_sales_order

@dp.materialized_view(name="sales_order_line")
@dp.expect_all_or_drop(EXPECTATIONS)
def sales_order_line():
    return to_silver_sales_order(
        spark.read.table("bronze.sap_vbap"),
        spark.read.table("silver.fx_rate"),
    )

import sys

# Both are job parameters set by the bundle, not relative paths and not
# an environment sniff.
source_root = dbutils.widgets.get("source_root")
catalog = dbutils.widgets.get("catalog")
if source_root not in sys.path:
    sys.path.insert(0, source_root)

from dwh.silver.customer import clean_customer   # noqa: E402

def run(spark, source: str, target: str) -> int:
    df = clean_customer(spark.read.table(source))
    df.write.mode("overwrite").saveAsTable(target)
    return df.count()

# The entry point is the only place a table name is assembled.
rows = run(spark, f"{catalog}.bronze.sap_kna1", f"{catalog}.silver.customer")
print(f"silver.customer: {rows} rows written")

resources:
  jobs:
    silver_customer:
      name: dwh-silver-customer
      tasks:
        - task_key: silver_customer
          notebook_task:
            notebook_path: ../src/dwh/jobs/silver_customer.py
            source: WORKSPACE
            base_parameters:
              source_root: ${workspace.file_path}/src
              catalog: ${var.catalog}

CREATE TABLE dwh_test.fixtures.char_fct_order_line_20260729
COMMENT 'Characterization snapshot of gold.fct_order_line, June 2026, taken before the transform refactor. Drop when the refactor is merged.'
AS
SELECT order_line_sk, order_id, line_number,
       customer_sk, product_sk, date_sk, region_sk,
       order_ts, quantity, unit_price, currency, fx_rate, amount_eur, status
       -- _tech_loaded_at is deliberately NOT projected. It changes on every
       -- run, so including it makes every row differ and the test useless.
FROM   dwh_prod.gold.fct_order_line
WHERE  order_ts >= DATE '2026-06-01'
  AND  order_ts <  DATE '2026-07-01';

-- Same projection AND same window on both sides, or every row outside
-- June is reported as a difference and the test can never return zero.
WITH new_rows AS (
  SELECT order_line_sk, order_id, line_number,
         customer_sk, product_sk, date_sk, region_sk,
         order_ts, quantity, unit_price, currency, fx_rate, amount_eur, status
  FROM   dwh_test.gold.fct_order_line
  WHERE  order_ts >= DATE '2026-06-01'
    AND  order_ts <  DATE '2026-07-01'
)
SELECT 'new_not_in_old' AS side, d.* FROM (
  SELECT * FROM new_rows
  EXCEPT ALL
  SELECT * FROM dwh_test.fixtures.char_fct_order_line_20260729
) d
UNION ALL
SELECT 'old_not_in_new' AS side, d.* FROM (
  SELECT * FROM dwh_test.fixtures.char_fct_order_line_20260729
  EXCEPT ALL
  SELECT * FROM new_rows
) d;

bundle:
  name: dwh

include:
  - resources/*.yml

variables:
  catalog:
    description: Unity Catalog catalog this target writes to.
  deploy_sp:
    description: Application ID of the OAuth service principal that owns prod runs.

targets:
  dev:
    mode: development
    default: true
    variables:
      catalog: dwh_dev

  test:
    variables:
      catalog: dwh_test
    resources:
      jobs:
        ci_assert: &ci_assert
          name: dwh-ci-assert-${bundle.target}
          tasks:
            - task_key: seed
              notebook_task:
                notebook_path: ../src/dwh/ci/seed.py
                source: WORKSPACE
                base_parameters:
                  source_root: ${workspace.file_path}/src
                  catalog: ${var.catalog}
            - task_key: pipeline
              depends_on:
                - task_key: seed
              pipeline_task:
                pipeline_id: ${resources.pipelines.silver.id}
                full_refresh: true
            - task_key: assert
              depends_on:
                - task_key: pipeline
              notebook_task:
                notebook_path: ../src/dwh/ci/assert_expectations.py
                source: WORKSPACE
                base_parameters:
                  source_root: ${workspace.file_path}/src
                  catalog: ${var.catalog}

  sandbox:
    variables:
      catalog: dwh_dev
    resources:
      jobs:
        ci_assert: *ci_assert     # referenced, not redefined

  prod:
    mode: production
    variables:
      catalog: dwh_prod
    run_as:
      service_principal_name: ${var.deploy_sp}
    # ci_assert is simply ABSENT here. That is the workaround.

name: dwh
on:
  pull_request:
    branches: [main]

jobs:
  verify:
    runs-on: ubuntu-latest
    env:
      DATABRICKS_HOST:          ${{ secrets.DATABRICKS_HOST }}
      DATABRICKS_CLIENT_ID:     ${{ secrets.DATABRICKS_CI_CLIENT_ID }}
      DATABRICKS_CLIENT_SECRET: ${{ secrets.DATABRICKS_CI_CLIENT_SECRET }}
    steps:
      - uses: actions/checkout@v4

      - uses: databricks/setup-cli@main
        with:
          version: 0.275.0          # pinned; see the bundles-python fact

      - name: bundle validate
        run: databricks bundle validate -t test

      # Catches uppercase AND non-ASCII in one pass, and only in the table
      # name — a COMMENT holding a German term is untouched.
      - name: identifier rule
        run: |
          if grep -rEn 'CREATE (OR REPLACE )?TABLE (IF NOT EXISTS )?[^ (]*[^a-z0-9._ (]' src/; then
            echo "table names must match ^[a-z][a-z0-9_]*$ and be ASCII" >&2
            exit 1
          fi

      - name: unit tests
        run: pytest tests/unit --junitxml=reports/unit.xml

      - name: deploy to test
        run: databricks bundle deploy -t test

      - name: seed, run, assert
        run: databricks bundle run -t test ci_assert