Lab: Build the Harbor Lane pipeline

'''Harbor Lane ELT: data/raw/*.csv -> staging views -> star schema.
Idempotent by construction: run it as many times as you like.'''
import time
import duckdb

con = duckdb.connect('harbor.duckdb')
t0 = time.time()

def step(label, sql):
    con.execute(sql)
    print(f'  ok  {label}')

# ---- staging: mechanical cleanup only, one view per source ----------------
step('stg_products', '''
  CREATE OR REPLACE VIEW stg_products AS
  SELECT sku, product_name, category, unit_cost::DECIMAL(8,2) AS unit_cost
  FROM read_csv_auto('data/raw/products.csv', header=true)
''')

step('stg_web_customers', '''
  CREATE OR REPLACE VIEW stg_web_customers AS
  SELECT web_id, lower(trim(email)) AS email, trim(first_name) AS first_name,
         trim(last_name) AS last_name, phone, zip, created_at::DATE AS created_at
  FROM read_csv_auto('data/raw/web_customers.csv', header=true)
''')

step('stg_pos_customers', '''
  CREATE OR REPLACE VIEW stg_pos_customers AS
  SELECT pos_id, trim(full_name) AS full_name,
         nullif(lower(trim(email)), '') AS email,
         phone, store_zip, signup_date::DATE AS signup_date
  FROM read_csv_auto('data/raw/pos_customers.csv', header=true)
''')

step('stg_orders', '''
  CREATE OR REPLACE VIEW stg_orders AS
  SELECT order_id, lower(channel) AS channel, customer_ref, sku,
         qty::INTEGER AS qty, unit_price::DECIMAL(8,2) AS unit_price,
         ordered_at::TIMESTAMP AS ordered_at,
         ordered_at::DATE AS order_date, lower(status) AS status
  FROM read_csv_auto('data/raw/orders_*.csv', header=true, union_by_name=true)
''')

# ---- dims: small, so full refresh - trivially idempotent ------------------
step('dim_product (full refresh)', '''
  CREATE OR REPLACE TABLE dim_product AS
  SELECT sku, product_name, category, unit_cost FROM stg_products
''')

# ---- fact: partition overwrite on order_date -------------------------------
step('fct_order_lines (create if needed)', '''
  CREATE TABLE IF NOT EXISTS fct_order_lines (
    order_id TEXT, channel TEXT, customer_ref TEXT, sku TEXT,
    qty INTEGER, unit_price DECIMAL(8,2), ordered_at TIMESTAMP,
    order_date DATE, status TEXT
  )
''')
step('fct_order_lines (delete batch dates)', '''
  DELETE FROM fct_order_lines
  WHERE order_date IN (SELECT DISTINCT order_date FROM stg_orders)
''')
step('fct_order_lines (insert batch)', '''
  INSERT INTO fct_order_lines
  SELECT order_id, channel, customer_ref, sku, qty, unit_price,
         ordered_at, order_date, status
  FROM stg_orders
''')

# ---- run log: every run leaves evidence ------------------------------------
con.execute('''
  CREATE TABLE IF NOT EXISTS pipeline_runs (
    ran_at TIMESTAMP DEFAULT now(), fact_rows BIGINT, max_order_date DATE,
    seconds DOUBLE
  )
''')
rows, hi = con.execute(
    'SELECT count(*), max(order_date) FROM fct_order_lines').fetchone()
con.execute('INSERT INTO pipeline_runs (fact_rows, max_order_date, seconds) VALUES (?, ?, ?)',
            [rows, hi, round(time.time() - t0, 3)])
print(f'\nfact rows: {rows}   watermark: {hi}   ({round(time.time() - t0, 2)}s)')

python3 pipeline.py     # first run:  fact rows: 3052, watermark: 2026-05-30
python3 pipeline.py     # second run: SAME fact rows, SAME watermark
python3 pipeline.py     # third run:  still the same. This is the whole point.