Full contract example | ContractForge

Silver SCD1 from JDBC

This contract reads a database table, applies deterministic deduplication, validates quality, evolves schema additively, writes current-state rows and records operations metadata.

preset: silver_scd1_upsert

source:
  type: connector
  connector: postgres
  options:
    url: "{{ secret:erp/postgres_url }}"
    dbtable: public.orders
  auth:
    type: basic
    username: "{{ secret:erp/user }}"
    password: "{{ secret:erp/password }}"
  read:
    partitionColumn: order_id
    lowerBound: 1
    upperBound: 5000000
    numPartitions: 8
    fetchsize: 10000

target:
  catalog: main
  schema: sales_curated
  table: s_orders

layer: silver
source_system: erp
mode: scd1_upsert
merge_keys: [order_id]
watermark_columns: [updated_at]
dedup_order_expr: "updated_at DESC NULLS LAST, order_id DESC NULLS LAST"

column_mapping:
  id: order_id
  customer: customer_id

transform:
  shape:
    columns:
      order_id:
        source: order_id
        cast: string
      customer_id:
        source: customer_id
        cast: string
      order_total:
        source: total_amount
        cast: decimal(18,2)
      updated_at:
        source: updated_at
        cast: timestamp

schema_policy: additive_only
allow_type_widening: true

delta_properties:
  delta.enableChangeDataFeed: "true"
  delta.autoOptimize.optimizeWrite: "true"
cluster_columns: [customer_id, updated_at]
optimize_after_write: true

quality_rules:
  required_columns: [order_id, customer_id, updated_at]
  not_null: [order_id, updated_at]
  unique_key: [order_id]
  min_rows: 1
  expressions:
    - name: non_negative_total
      expression: "order_total >= 0"
      severity: quarantine
      message: "Order total must not be negative."

on_quality_fail: quarantine
idempotency_key: "orders:${run_date}"
idempotency_policy: skip_if_success
retry_attempts: 5
retry_backoff_seconds: 10
lock_enabled: true

operations:
  business_owner: sales-ops
  technical_owner: data-platform
  support_group: data-platform
  criticality: high
  expected_frequency: daily
  freshness_sla_minutes: 180
  runbook_url: https://wiki.example.com/runbooks/orders

annotations:
  table:
    description: Curated current-state orders table.
    tags:
      domain: sales
      layer: silver
  columns:
    customer_id:
      description: Business customer identifier.
      tags:
        entity: customer

from contractforge import ingest

result = ingest(
    preset="silver_scd1_upsert",
    source={
        "type": "connector",
        "connector": "postgres",
        "options": {
            "url": "{{ secret:erp/postgres_url }}",
            "dbtable": "public.orders",
        },
        "auth": {
            "type": "basic",
            "username": "{{ secret:erp/user }}",
            "password": "{{ secret:erp/password }}",
        },
        "read": {
            "partitionColumn": "order_id",
            "lowerBound": 1,
            "upperBound": 5_000_000,
            "numPartitions": 8,
            "fetchsize": 10_000,
        },
    },
    target={"catalog": "main", "schema": "sales_curated", "table": "s_orders"},
    layer="silver",
    source_system="erp",
    mode="scd1_upsert",
    merge_keys=["order_id"],
    watermark_columns=["updated_at"],
    dedup_order_expr="updated_at DESC NULLS LAST, order_id DESC NULLS LAST",
    column_mapping={"id": "order_id", "customer": "customer_id"},
    transform={"shape": {"columns": {"order_total": {"source": "total_amount", "cast": "decimal(18,2)"}}}},
    schema_policy="additive_only",
    allow_type_widening=True,
    quality_rules={"not_null": ["order_id", "updated_at"], "unique_key": ["order_id"]},
    on_quality_fail="quarantine",
    idempotency_key="orders:${run_date}",
    idempotency_policy="skip_if_success",
)

How to read this example

Block	Purpose
`source`	Defines only how data is read. Business shaping happens later.
`target`	Defines the physical destination. `layer` remains operational metadata.
`transform.shape`	Projects and normalizes nested or renamed fields before quality and write.
`quality_rules`	Defines gates for safe writes and quarantine-capable expressions.
`operations`	Supplies ownership and SLA metadata for dashboards and alerts.
`annotations`	Documents the resulting table and fields in the catalog.

Keep contracts intentional

Do not copy every field into every contract. Use presets and templates for defaults, then keep only fields that communicate a decision reviewers should understand.