Reference: A Data Platform / ETL System

data-platform/
  flows/ingest_orders.py     # orchestrated flow
  transforms/orders.py       # pure functions
  loaders/warehouse.py       # COPY / MERGE
  contracts/orders.yaml      # schema + SLAs
  tests/test_transforms.py

"""transforms/orders.py - idempotent transform."""
from __future__ import annotations
 
import polars as pl
 
 
def normalize_orders(raw: pl.LazyFrame) -> pl.LazyFrame:
    return (
        raw.filter(pl.col("status").is_in(["paid", "refunded"]))
        .with_columns(
            pl.col("amount").cast(pl.Float64),
            pl.col("created_at").str.to_datetime(strict=False),
        )
        .select("order_id", "tenant_id", "amount", "status", "created_at")
    )

"""flows/ingest_orders.py - Prefect-style flow sketch."""
from __future__ import annotations
 
from datetime import datetime, timezone
 
import polars as pl
 
from loaders.warehouse import merge_into_warehouse
from transforms.orders import normalize_orders
 
WATERMARK_KEY = "orders_ingest"
 
 
def run_ingest(since: datetime) -> int:
    raw_path = f"s3://lake/raw/orders/{since.date()}.parquet"
    raw = pl.scan_parquet(raw_path)
    clean = normalize_orders(raw).collect()
    rows = merge_into_warehouse(clean, table="analytics.orders", key="order_id")
    return rows
 
 
if __name__ == "__main__":
    ts = datetime.now(timezone.utc)
    print(f"loaded {run_ingest(ts)} rows")

# contracts/orders.yaml
name: orders
freshness_sla_hours: 6
columns:
  order_id: { type: string, required: true }
  amount: { type: float, min: 0 }

# loaders/warehouse.py sketch
def merge_into_warehouse(df: pl.DataFrame, table: str, key: str) -> int:
    # write parquet staging → MERGE SQL
    return df.height