Code structure

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"),
    )

from pyspark.sql import SparkSession

from dwh.silver.customer import clean_customer

def run(spark: SparkSession, source: str, target: str) -> int:
    """Fully-qualified names in, row count out. Nothing here reads a widget,
    so pytest can import this module on a laptop."""
    df = clean_customer(spark.read.table(source))
    df.write.mode("overwrite").saveAsTable(target)
    return df.count()

# Databricks notebook source
# ^ that first line is not a comment you can drop: without it the task is a
#   plain .py file and the CLI refuses it. See the callout below.
import sys

# Both are job parameters set by the bundle, not relative paths and not
# an environment sniff. dbutils and spark are notebook globals — they exist
# here and nowhere under src/dwh/jobs/ or src/dwh/silver/.
source_root = dbutils.widgets.get("source_root")   # noqa: F821
catalog = dbutils.widgets.get("catalog")           # noqa: F821
if source_root not in sys.path:
    sys.path.insert(0, source_root)

from dwh.jobs.silver_customer import run           # noqa: E402

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

resources:
  jobs:
    silver_customer:
      name: dwh-silver-customer
      tasks:
        - task_key: silver_customer
          notebook_task:
            # Relative to THIS file (resources/jobs.yml), so ../ is correct
            # here and wrong in databricks.yml. The file starts with
            # "# Databricks notebook source".
            notebook_path: ../src/dwh/jobs/entry/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;