Data contracts that execute, not just document.

"""
E-Commerce Orders → silver layer.
 
Executes the contract end-to-end: schema + PII + row & dataset quality rules.
Bad rows are quarantined with reasons; good rows promoted to silver.
Same contract runs on DuckDB, Spark, or Polars.
"""
 
# pip install lakelogic
 
import lakelogic as ll
 
contract = """
version: 1.0.0
dataset: orders
 
info:
  title: E-Commerce Orders
  owner: data-team@company.com
  target_layer: silver
  domain: sales
  system: shopify
 
model:
  fields:
    - name: order_id
      type: integer
      description: "Unique numeric ID for each order"
    - name: customer_email
      type: string
      pii: true
      description: "Customer's raw email — flagged PII for governance"
    # - name: customer_email_formatted
    #   type: string
    #   pii: true
    - name: customer_email_encrypted
      type: string
      pii: true
      masking: partial
      description: "Same email, partially masked at runtime (r***@latezuab.co)"
    - name: amount
      type: float
      required: true
      description: "Order total in USD; row-rules guard against zero / negative"
    - name: status
      type: string
      description: "Order lifecycle: pending · shipped · delivered"
 
transformations:
  - sql: "SELECT *, UPPER(LTRIM(customer_email)) as customer_email_formatted
            FROM source"
 
quality:
  row_rules:
    - name: positive_amount
      sql: "amount > 0"
    - name: valid_status
      sql: "status IN ('pending','shipped','delivered')"
 
  dataset_rules:
    - name: minimum_total_revenue
      description: "Batch must carry at least $1,000 in valid sales"
      sql: "SELECT SUM(amount) FROM {dataset}"
      must_be_greater_than: 1000
 
    - name: max_pending_ratio
      description: "Pending orders capped at 50% of the batch"
      sql: "SELECT COUNT(*) FILTER(WHERE status='pending')*1.0
              / COUNT(*) FROM {dataset}"
      must_be_less_than: 0.5
 
# ── Optional: Physical Execution Config ────────────────────────────
# source:
#   type: delta
#   path: "abfss://landing@myaccount.dfs.core.windows.net/orders"
#   load_mode: incremental
#   watermark_strategy: pipeline_log
#
# materialization:
#   path: "abfss://silver@myaccount.dfs.core.windows.net/orders"
#   strategy: merge
#   format: delta
#   merge_dedup_guard: true
#   partition_by:
#     - status
 
# ── Observatory Telemetry — OSS → LakeLogic Cloud bridge ──────────
# observatory:
#   enabled: false
#   environments: ["dev", "prod", "staging", "local"]
#   endpoint: "${LAKELOGIC_OBSERVATORY_ENDPOINT}"
#   emit_on: [success, partial, failed]
#   expected_delivery:                  # Telemetry Delivery Watchdog
#     cadence_minutes: 15
#     missing_after_minutes: 45
"""
 
# Generate test data — 50% intentionally bad
source_df = ll.DataGenerator(contract).generate(rows=1000, invalid_ratio=0.5)
 
# Run the pipeline
proc = ll.DataProcessor(contract, engine="duckdb")
good, bad = proc.run(source_df)