Zero to First Audit: Bringing an Apache Airflow Deployment under CoreModels Governance

Your Airflow deployment can already describe itself. The stable REST API v1 will tell you every DAG, every task, and - if you use data-aware scheduling - every asset your pipelines produce and consume. What it will not tell you is whether any of that still matches what your organization thinks it runs: who owns each pipeline, which assets are consumed but produced by nothing, which consumers are quietly running on stale data because every producer is paused.

That gap is what a CoreModels audit closes. In this tutorial we go from three curl calls against your own Airflow API to a governed model of your orchestration estate and a first drift audit - no plugins installed in Airflow, no credentials shared with us, nothing running inside your deployment.

A note on capabilities before we start: the Airflow connector (vendor key airflow) supports Import and Audit. It deliberately does not support Generate - orchestration code is not derivable from schema, and the connector refuses honestly rather than emitting something half-true.

Step 1: Extract the artifacts

CoreModels never connects to your Airflow instance. You call your own stable REST API v1 and upload the JSON responses as artifacts. Three artifacts exist; only the first is required.

Artifact	Source	Required?
dags	GET /api/v1/dags - the full response or just its dags array	yes
tasks	GET /api/v1/dags/{dag_id}/tasks, aggregated as {"<dag_id>": <response>}	no
datasets	GET /api/v1/datasets - data-aware scheduling assets ("Assets" in Airflow 3)	no

The exact recipe:

AIRFLOW=https://your-airflow.example.com
AUTH='-u your-user:your-password'   # or -H "Authorization: Bearer <token>"

curl -s $AUTH "$AIRFLOW/api/v1/dags?limit=200" > dags.json

for d in $(jq -r '.dags[].dag_id' dags.json); do
  curl -s $AUTH "$AIRFLOW/api/v1/dags/$d/tasks" | jq --arg d "$d" '{($d): .}'
done | jq -s 'add' > tasks.json

curl -s $AUTH "$AIRFLOW/api/v1/datasets" > datasets.json

Skip tasks.json and you still get governed DAGs and lineage, just without per-task detail. Skip datasets.json and you lose cross-DAG lineage - the most valuable part if you use data-aware scheduling - so include it if you can.

Step 2: Import into a CoreModels project

You need a CoreModels project (its 32-character hex id is $PROJECT_ID below) and an Admin role on it, because import writes to the graph. It writes additively: existing governed nodes are never mutated, so re-importing later is safe.

Artifact values are the raw file contents passed as JSON strings, which is exactly what jq --rawfile produces:

jq -n --rawfile dags dags.json --rawfile tasks tasks.json --rawfile ds datasets.json \
  '{artifacts: {dags: $dags, tasks: $tasks, datasets: $ds}}' > import-request.json

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

A successful import answers with counts and two honesty channels:

{ "success": true, "vendor": "airflow", "projectName": "airflow",
  "datasetsAdded": 12, "datasetsSkippedExisting": 0, "fieldsAdded": 0,
  "lineageEdgesAdded": 9, "lineageEdgesSkipped": 0, "nodesEnriched": 60,
  "snapshotStored": true,
  "lossiness": [
    { "kind": "TypeApproximation", "path": "daily_revenue.build_report",
      "explanation": "Native type 'PythonOperator' was approximated as String; the exact native type is preserved in the vendor metadata mixin." }
  ],
  "errors": [] }

Read this as: each DAG and each data-aware-scheduling asset became a governed Type (datasetsAdded); each task became an Element on its DAG, with the operator class recorded as its native type - new DAGs carry their tasks in within datasetsAdded, while fieldsAdded counts only tasks added to already-governed DAGs on a later re-import, so it is 0 on a first import; each producing/consuming relationship became a lineage edge in the same Depends On graph that warehouse and dbt estates use (lineageEdgesAdded). The lossiness array is where the import confesses anything it approximated or dropped - for Airflow it carries one type-approximation record per task, because an operator class is not a data type - and errors is empty unless it could not proceed at all. snapshotStored: true means the parsed deployment snapshot was persisted, which enables one-call re-audits later without fresh artifacts.

Step 3: Run your first audit

The audit compares fresh artifacts against the governed model. It is strictly read-only and needs only the Viewer role; setting recordHistory: true opts the run into the rolling drift trail:

jq -n --rawfile dags dags.json --rawfile ds datasets.json \
  '{artifacts: {dags: $dags, datasets: $ds}, recordHistory: true}' > audit-request.json

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

Run immediately after an import, the drift sections should be clean - you just imported this exact estate. What you will almost certainly see on a real deployment is conformance findings, because the connector ships four Airflow-specific hygiene rules.

Step 4: Read the result

Start with the counts:

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

{
  "errorCount": 0,
  "warningCount": 3,
  "infoCount": 2,
  "codes": { "dag-no-owner": 2, "dag-no-description": 2, "paused-producer": 1 }
}

errorCount is the machine signal - greater than zero means the estate violates governed meaning, and it is what a CI gate keys on. Warnings and infos are hygiene. Each finding in the findings array carries a section (Coverage, Drift, or Conformance), a severity, a stable code, a subject, and a human message. Here is a real one:

{ "section": "Conformance", "severity": "Warning", "code": "dag-no-owner",
  "subject": "daily_revenue",
  "message": "DAG has no real owner (default 'airflow') - nobody is accountable for this pipeline." }

The four Airflow-specific rules are worth knowing on day one:

Code	Severity	What it catches
dag-no-description	Info	a DAG without stated intent
dag-no-owner	Warning	owner missing or the default airflow - nobody accountable
asset-unproduced	Warning	an asset DAGs consume but nothing in this deployment produces
paused-producer	Warning	every producer of an asset paused while active DAGs still consume it - silent staleness

On top of those, the shared audit engine reports coverage (dataset-unmapped, field-unmapped) and drift (dataset-removed, field-removed, field-type-drift, contract-drift, and enum-change codes) once your estate and governed model start evolving apart.

The response also includes markdown - a complete, PR-comment-ready report - so a human never has to read the JSON:

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

And because you passed recordHistory: true, this run is now the first entry in the project's audit trail. Two quick follow-ups you can try right away, both Viewer-role GETs:

curl -sS -H "Authorization: Bearer $TOKEN" \
  "https://coremodels.example.com/graph/integrations/airflow/status/$PROJECT_ID"

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

status returns the last-import state (imported, state, governedDatasets); badge returns a shields-style SVG rendered from the latest recorded run - green for clean, yellow for warnings, red for errors, gray when nothing is recorded yet.

Where this goes next

You now have a governed model of your orchestration estate and a repeatable audit. From here the same three files plug into a CI drift gate on your DAG repository (the machine-to-machine v1 audit route accepts user API keys), the stored snapshot enables one-call re-audits whenever the governed model changes, and every governed fact - ownership, schedules, the full cross-DAG dependency graph - is queryable by AI agents over MCP. Each of those is a tutorial of its own.

One honest limitation to keep in mind: on very large deployments, when the encoded snapshot exceeds the storage cap (~1.5 MB), import reports snapshotStored: false with a lossiness record. Fresh-artifact audits keep working; only the stored-snapshot re-audit is unavailable.

For the condensed version of everything above, see the Apache Airflow quickstart in the CoreModels integration docs.