Full contract example.

Use this page as a map for the ingestion contract surface. The example is intentionally broad, but production contracts should only include fields that express a real decision.

Silver current-state from JDBC

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

YAML

preset: silver_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: upsert
merge_keys: [order_id]
watermark_columns: [updated_at]

column_mapping:
  id: order_id
  customer: customer_id

transform:
  shape:
    columns:
      order_id:
        cast: string
      customer_id:
        cast: string
      total_amount:
        alias: order_total
        cast: decimal(18,2)
      updated_at:
        cast: timestamp
  cast:
    order_id: string
    customer_id: string
    order_total: decimal(18,2)
    updated_at: timestamp
  standardize:
    customer_id:
      trim: true
      empty_as_null: true
  derive:
    updated_date: to_date(updated_at)
  deduplicate:
    keys: [order_id]
    order_by:
      - column: updated_at
        direction: desc
        nulls: last
      - column: order_id
        direction: desc
        nulls: last

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

Python

from contractforge import ingest

result = ingest(
    preset="silver_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="upsert",
    merge_keys=["order_id"],
    watermark_columns=["updated_at"],
    column_mapping={"id": "order_id", "customer": "customer_id"},
    transform={
        "shape": {"columns": {"total_amount": {"alias": "order_total", "cast": "decimal(18,2)"}}},
        "cast": {
            "order_id": "string",
            "customer_id": "string",
            "order_total": "decimal(18,2)",
            "updated_at": "timestamp",
        },
        "standardize": {"customer_id": {"trim": True, "empty_as_null": True}},
        "derive": {"updated_date": "to_date(updated_at)"},
        "deduplicate": {
            "keys": ["order_id"],
            "order_by": [
                {"column": "updated_at", "direction": "desc", "nulls": "last"},
                {"column": "order_id", "direction": "desc", "nulls": "last"},
            ],
        },
    },
    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	Projects nested data, casts values, standardizes text, derives columns and deduplicates 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.