Exemplo de contrato completo.

Use esta página como um mapa para a superfície do contrato de ingestão. O exemplo é intencionalmente amplo, mas os contratos de produção devem incluir apenas campos que expressem uma decisão real.

Prata SCD1 da JDBC

Este contrato lê uma tabela de banco de dados, aplica desduplicação determinística, valida a qualidade, evolui o esquema de forma aditiva, grava linhas de estado atual e registra metadados de operações.

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",
)

Como ler este exemplo

Bloquear	Propósito
source	Define apenas como os dados são lidos. A formação dos negócios acontece mais tarde.
target	Define o destino físico. layer permanecem metadados operacionais.
transform	Projeta dados aninhados, converte valores, padroniza texto, deriva colunas e desduplica antes da qualidade e gravação.
quality_rules	Define portas para gravações seguras e expressões com capacidade de quarentena.
operations	Fornece metadados de propriedade e SLA para painéis e alertas.
annotations	Documenta a tabela e os campos resultantes no catálogo.

Mantenha os contratos intencionais

Não copie todos os campos em todos os contratos. Use presets e modelos para padrões e mantenha apenas os campos que comunicam uma decisão que os revisores devem entender.