Enforce OpenMetadata Catalog Contracts at Ingestion with DataContractGate

OpenMetadata is where the catalog, owners, and the Data Quality tab live. DataContractGate is the gate in front of the table. Export the contract as ODCS v3.1, enforce it on every event, and write the pass and quarantine counts back as a test result.

1. Export the contract from OpenMetadata

OpenMetadata stores a data contract on a table and can export it as ODCS YAML. The integration reads that document with the catalog API:

GET /v1/dataContracts/{id}/odcs/yaml
Accept: application/yaml
Authorization: Bearer $OM_JWT

The document should be ODCS apiVersion: v3.1.0 and kind: DataContract. Field names, logical types, required flags, descriptions, and allowed values expressed as invalidValues / validValues travel. Draft one in the playground if you are starting from a sample payload and then registering it in the catalog.

2. Import it into DataContractGate and enforce it live

The publisher package lives in the DataContractGate repo at integrations/openmetadata. Contract sync pulls the ODCS document and posts it to DataContractGate:

export OM_HOST=https://openmetadata.example
export OM_JWT=...
export DCG_API_URL=https://api.datacontractgate.com
export DCG_API_KEY=cg_live_...

python -m dcg_openmetadata sync --om-contract-id <openmetadata-contract-uuid>
# new draft version of a contract you already run:
python -m dcg_openmetadata sync \
  --om-contract-id <openmetadata-contract-uuid> \
  --target-contract-id <datacontractgate-uuid>

Import creates a draft with requires_review: true. Read the draft, tighten anything the catalog dropped, and promote a stable version. Until that promotion, ingest keeps using the previous stable contract. Producers then call POST /v1/ingest/{contract_id}. Failing events are quarantined. Passing events are the ones that may be loaded into the table the catalog describes.

3. Publish pass and quarantine results to the Data Quality tab

On a schedule, the publisher reads GET /contracts/{id}/report and writes one OpenMetadata test result. The first run creates a custom test definition named dcgIngestContract and one test case on the mapped table. Later runs only add results.

{
  "window_minutes": 60,
  "skip_empty": true,
  "defaults": { "max_violation_rate": 0.0 },
  "contracts": [
    {
      "contract_id": "<datacontractgate-uuid>",
      "table_fqn": "analytics.warehouse.public.orders",
      "test_case_name": "dcg_orders_ingest",
      "max_violation_rate": 0.01
    }
  ]
}

python -m dcg_openmetadata publish --config mapping.json --dry-run
python -m dcg_openmetadata publish --config mapping.json

Each result carries:

• testCaseStatus — Failed when quarantined ÷ total is above max_violation_rate, otherwise Success.
• testResultValue — total, passed, quarantined, violation_rate, contract_version.
• passedRows and failedRows — events that passed, and events that were quarantined.
• A one-line summary with the top violation kinds, for example amount:range_violation.

Point the test case at the table with table_fqn (service.database.schema.table), or put the fully qualified name on the contract itself:

x-openmetadata-table: analytics.warehouse.public.orders

Schedule publish at the same interval as window_minutes. The Data Quality tab, the test case page, and any OpenMetadata incident rule then show the ingest gate next to the warehouse tests. Map a contract to a table once. Empty windows are skipped when skip_empty is true, which is the default.

What each side owns

Source and the walkthrough: integrations/openmetadata. Environment variables are OM_HOST, OM_JWT, DCG_API_URL, DCG_API_KEY, and, for a local dev gateway only, DCG_ORG_ID. Keep the JWT and the cg_live_… key in the environment, not in the mapping file.

See also how this gate sits in front of warehouse tests and how the same ODCS v3.1 document runs in CsvPath.