Zero to First Audit: Putting an Airbyte Catalog Under Governance

Everything in this tutorial happens in a shell. There is no agent to install in your Airbyte deployment, no warehouse connection to configure, and no credential of yours that ever reaches CoreModels. You export one JSON document that Airbyte already produces, upload it, and ask a question about it.

By the end you will have an Airbyte estate represented as governed meaning in a CoreModels project, and a first drift audit that tells you — in a single integer — whether a fresh catalog still matches what your organization agreed that data means.

Throughout: https://coremodels.example.com stands in for your CoreModels API base URL, $TOKEN is your login token, and $PROJECT_ID is the 32-character hex id of the governing project. Import requires the Admin role on that project; the audit needs only Viewer.

The connector wants exactly one artifact

The Airbyte connector is registered under the vendor key airbyte and declares one artifact:

{
  "key": "airbyte",
  "displayName": "Airbyte",
  "capabilities": "Import, Audit",
  "artifacts": {
    "catalog": "required — an AirbyteCatalog (source discover output) or ConfiguredAirbyteCatalog (connection export) JSON"
  }
}

That is the live response fragment from GET graph/integrations/vendors, so it is also your capability contract: Airbyte supports Import and Audit, and deliberately not Generate. A catalog describes what a source exposes; authoring one belongs to the source, not to the governed model.

Both catalog shapes parse. The parser only insists on {"streams": [...]} with at least one stream, and — for a configured catalog — it reads the per-stream configuration wrapper as well as the nested stream object.

Step 1 — Export the catalog

AIRBYTE=https://airbyte.internal.example.com

# Option A — source discovery: what the source CAN expose.
curl -s -X POST "$AIRBYTE/api/v1/sources/discover_schema" \
  -H "Content-Type: application/json" \
  -d '{"sourceId": "<SOURCE_ID>"}' | jq '.catalog' > catalog.json

# Option B — a connection's configured catalog: what you actually sync,
# including sync modes, cursor fields and configured primary keys.
curl -s -X POST "$AIRBYTE/api/v1/web_backend/connections/get" \
  -H "Content-Type: application/json" \
  -d '{"connectionId": "<CONNECTION_ID>"}' | jq '.syncCatalog' > catalog.json

Add whatever auth headers your Airbyte deployment expects — they stay on your side of the wire. Prefer Option B when you have the choice: the configured catalog carries the operational decisions (incremental versus full refresh, cursor field, configured primary key) that the audit's conformance rules look at.

Sanity-check the export before uploading it:

jq '{streams: (.streams | length),
     names: [.streams[] | (.stream // .) | ((.namespace // "") + "." + .name)]}' catalog.json

For the rest of this tutorial we use a small three-stream catalog — public.users, events, public.order_items — with eleven properties between them. Here is the first of those three streams; substituting your own numbers is the point of the exercise.

{
  "streams": [
    {
      "stream": {
        "name": "users",
        "namespace": "public",
        "json_schema": {
          "type": "object",
          "properties": {
            "id": { "type": "integer", "description": "User key." },
            "email": { "type": ["null", "string"], "format": "email" },
            "status": { "type": "string", "enum": ["active", "suspended", "deleted"] },
            "created_at": {
              "type": ["null", "string"],
              "format": "date-time",
              "airbyte_type": "timestamp_with_timezone"
            }
          },
          "required": ["id", "status"]
        },
        "supported_sync_modes": ["full_refresh", "incremental"],
        "source_defined_cursor": true,
        "default_cursor_field": ["created_at"],
        "source_defined_primary_key": [["id"]]
      },
      "sync_mode": "incremental",
      "destination_sync_mode": "append_dedup",
      "primary_key": [["id"]],
      "cursor_field": ["created_at"]
    }
  ]
}

Step 2 — Import

The request body carries the raw file contents as a JSON string, which is exactly what jq --rawfile produces. Building the body in a file also keeps a multi-megabyte catalog out of your shell's argument list.

jq -n --rawfile catalog catalog.json '{artifacts: {catalog: $catalog}}' > import-request.json

curl -sS -X POST \
  "https://coremodels.example.com/graph/integrations/airbyte/import/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  --data-binary @import-request.json | jq

{
  "success": true,
  "vendor": "airbyte",
  "projectName": "airbyte-connection",
  "datasetsAdded": 3,
  "datasetsSkippedExisting": 0,
  "fieldsAdded": 0,
  "lineageEdgesAdded": 0,
  "lineageEdgesSkipped": 0,
  "nodesEnriched": 14,
  "snapshotStored": true,
  "lossiness": [
    {
      "kind": "TypeApproximation",
      "path": "public.users.email",
      "explanation": "Native type 'email' was approximated as String; the exact native type is preserved in the vendor metadata mixin."
    }
  ],
  "errors": []
}

(One lossiness record is shown; the events stream's payload property produces a second one, for the same reason.)

Four numbers are worth understanding before you move on.

datasetsAdded: 3 — one governed Type per stream. fieldsAdded: 0 is not a mistake: that counter reports fields added to already-governed datasets on a re-import. On a first import the fields arrive with their new Types. nodesEnriched: 14 is every node that received Airbyte metadata — three streams plus eleven properties. snapshotStored: true means the parsed estate was persisted, which is what makes the artifact-free re-audit possible later.

projectName is always airbyte-connection. A catalog carries no project name of its own, so the connector uses a stable estate name; you will need it if you ever want to target a specific stored snapshot.

The lossiness array is a success channel, not an error channel. Here it is telling you that email is a JSON Schema format, not a type the governed model has a primitive for, so the element became a String while the exact string email was preserved on the Airbyte metadata mixin. Nothing was dropped silently.

What landed in the graph: each stream became a Type with namespace-qualified identity (public.users); each property became an Element; the closed value set on status became a Taxonomy with the terms active, suspended, deleted; required entries and primary keys became not-null and unique checks; and the sync modes, cursor field and destination sync mode were recorded as metadata rather than invented as governed meaning.

Step 3 — The first audit

Audit compares a fresh catalog against the live governed graph. Run it against the same file you just imported and you get a clean coverage baseline plus whatever the estate's own hygiene rules find.

jq -n --rawfile catalog catalog.json \
  '{artifacts: {catalog: $catalog}, recordHistory: false}' > audit-request.json

curl -sS -X POST \
  "https://coremodels.example.com/graph/integrations/airbyte/audit/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  --data-binary @audit-request.json > audit-response.json

recordHistory defaults to false — the audit verb stays strictly read-only unless you ask it to leave a trace.

Step 4 — Read the result

jq '{errorCount, warningCount, infoCount, codes, driftedObjects, fingerprint, metrics}' audit-response.json

{
  "errorCount": 0,
  "warningCount": 1,
  "infoCount": 2,
  "codes": {
    "stream-no-primary-key": 1,
    "no-cursor-field": 1,
    "untyped-fields": 1
  },
  "driftedObjects": [],
  "fingerprint": "9f2c41ab7d0e5b83",
  "metrics": {
    "datasets (estate)": "3",
    "datasets governed": "3 / 3",
    "fields governed": "11 / 11",
    "governed nodes with canonical mappings": "0 / 15 (0%)",
    "last import": "2026-08-04T09:12:44.1183920+00:00"
  }
}

Read it in this order:

• errorCount is the gate. Zero means nothing in this catalog violates governed meaning. It is the one number automation needs.
• driftedObjects is empty, so nothing moved relative to the governed model. When it is non-empty it names the exact stream and field identities that drifted.
• codes is the compact machine summary. All three findings here are conformance rules the Airbyte connector contributes: one stream with no primary key (Warning), one stream that supports incremental sync with no cursor configured (Info), and one stream with object or untyped properties (Info).
• metrics gives you coverage at a glance. datasets governed: 3 / 3 and fields governed: 11 / 11 mean the estate and the graph agree. governed nodes with canonical mappings counts nodes that also carry a non-vendor mapping — a standards or ontology binding — which is 0 until someone adds one.

The individual findings carry the detail:

jq -r '.findings[] | "\(.severity)\t\(.code)\t\(.subject)\t\(.message)"' audit-response.json

Warning	stream-no-primary-key	events	Stream declares no primary key — dedup-dependent sync modes and downstream identity break silently.
Info	no-cursor-field	events	Stream supports incremental sync but no cursor field is defined — full refreshes will be the default.
Info	untyped-fields	events	1 field(s) are untyped or semi-structured — their inner schema enters the warehouse ungoverned.

For a human, the same report is already rendered:

jq -r '.markdown' audit-response.json

It opens with a verdict line — 🟡 **0 errors · 1 warnings · 2 info** — followed by the metrics table and one section per report area (Coverage, Drift, Conformance): a collapsible block when the area has findings, a one-line "no findings" otherwise. It is designed to be pasted straight into a pull request.

Step 5 — Make it repeatable

Two small additions turn a one-off check into a running signal. Set recordHistory: true on an audit and the run is appended to the project's rolling trail; then the badge endpoint renders the latest recorded run as an SVG:

curl -sS "https://coremodels.example.com/graph/integrations/airbyte/history/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN" | jq '.projects[].runs[0]'

curl -sS "https://coremodels.example.com/graph/integrations/airbyte/badge/$PROJECT_ID" \
  -H "Authorization: Bearer $TOKEN" -o airbyte-audit.svg

Green means clean, yellow means warnings only, red carries the error count, gray means nothing has been recorded yet.

Two things to know before you scale this up

Very large catalogs. The parsed estate snapshot is stored under a size cap. When the encoded snapshot exceeds it, import returns snapshotStored: false together with a lossiness record saying so. Audits with fresh artifacts keep working exactly as before; only the artifact-free re-audit has nothing to run against.

Generate is refused, on purpose. Call the generate route for airbyte and you get {"success": false, ... "errors": [{"path": "capabilities", "message": "Connector 'airbyte' does not support generation."}]}. That refusal is the same honesty contract as lossiness and snapshotStored: we would rather tell you a capability is absent than emit something that looks authoritative and is not.

From here, the natural next step is putting the audit call in front of your source-schema changes rather than behind them. The Airbyte quickstart in the CoreModels integration docs has the ready-made CI snippet.