# Airbyte — Quickstart: 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. # 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: ```json { "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 ```bash 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": ""}' | 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": ""}' | 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: ```bash 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. ```json { "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. ```bash 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 ``` ```json { "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. ```bash 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 ```bash jq '{errorCount, warningCount, infoCount, codes, driftedObjects, fingerprint, metrics}' audit-response.json ``` ```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: ```bash 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: ```bash 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: ```bash 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.