REST API, External Triggers, and Dynamic DAGs

How other systems start DAG runs, how conf and params meet, and how to generate many DAGs without making parsing expensive.

Most of this series has treated the bookshop pipeline as something Airflow decides to run — on a schedule, on an Asset update, on a sensor firing. But a lot of real orchestration starts outside Airflow. A CI job finishes building a new data image and wants to kick off a validation run. A finance app lets an analyst press “recompute Q3” and expects a pipeline to start with that quarter baked in. A sibling DAG finishes staging and needs to hand off to a marts DAG that another team owns. In every one of these, something that isn’t the scheduler needs to say “run this DAG, with this input, now.”

Airflow 3.x gives you three doors for that, and this chapter walks through all three: the stable REST API v2 for outside systems, the TriggerDagRunOperator for DAG-to-DAG handoffs, and the Param/conf contract that makes any triggered run safe to accept input. Then we close on the other kind of “dynamic” — generating many DAGs at parse time from a config file, and the parsing-cost tax that comes with it.

The stable REST API v2 is the front door

In Airflow 3.x the API story finally settled. The old experimental API is removed — if you have scripts hitting /api/experimental/..., they stop working on upgrade. In its place is a single stable REST API v2, served by the api-server (the same process that backs the React UI), documented with an OpenAPI spec, and versioned so you can build against it without fear of a point release moving the furniture. Everything the UI does, the API can do, because the UI is just a client of it.

Two things changed that trip people up. First, the base path is /api/v2/, not /api/v1/. Second, auth is token-based: you present a short-lived JWT on every request, and there is no more Basic-auth-against-a-web-user fallback baked into the core API. You obtain a token, then use it as a bearer credential until it expires.

Getting a token

The api-server exposes an auth endpoint that trades credentials for a JWT. With the default auth manager, that’s a POST with a username and password, and you get back a token you cache and reuse:

# 1. Trade credentials for a JWT
TOKEN=$(curl -s -X POST "http://localhost:8080/auth/token" \
  -H "Content-Type: application/json" \
  -d '{"username": "admin", "password": "admin"}' \
  | python -c "import sys, json; print(json.load(sys.stdin)['access_token'])")

echo "$TOKEN"   # eyJhbGciOiJIUzI1NiIsInR5cCI6...

That admin/admin pair is the Astro CLI local default — the same login you use for the UI. In a real deployment the credentials come from whatever auth manager you’ve configured, and for machine-to-machine access you’d issue a service identity rather than reusing a human’s login. The important part is the shape: authenticate once, get a JWT, send it as Authorization: Bearer <token> on every subsequent call. Tokens are short-lived by design, so a long-running client re-fetches when one expires rather than trying to make a single token immortal.

Triggering a run with conf

Here is the call the whole chapter orbits — start a DAG run and hand it a JSON conf payload:

curl -s -X POST \
  "http://localhost:8080/api/v2/dags/bookshop_reprocess/dagRuns" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
        "logical_date": "2026-06-24T00:00:00Z",
        "conf": {
          "quarter": "2026-Q2",
          "regions": ["us", "eu"],
          "full_refresh": true
        }
      }'

The response echoes the created run — its dag_run_id, state (queued), the logical_date, and the conf you passed:

{
  "dag_run_id": "manual__2026-06-24T00:00:00+00:00",
  "dag_id": "bookshop_reprocess",
  "logical_date": "2026-06-24T00:00:00Z",
  "state": "queued",
  "run_type": "manual",
  "conf": { "quarter": "2026-Q2", "regions": ["us", "eu"], "full_refresh": true }
}

That conf dict is the entire external input to the run. Inside the DAG, tasks read it back (we’ll see how in the params section), so conf is the wire format for “recompute Q2 for US and EU with a full refresh.” And logical_date is not optional, which catches everyone porting from 2.x. Omit the key entirely and the API rejects the request:

{"type": "missing", "loc": ["body", "logical_date"], "msg": "Field required"}

If you don’t care about the date — you just want a run now — you must say so explicitly by sending "logical_date": null. That reads like pedantry and isn’t: the whole point of Airflow 3 making it required is that “what data window is this run for?” should be an answer the caller gives, not one the server guesses. Passing an explicit date is what you want when the run is keyed to a business date; passing null is what you want when it genuinely isn’t.

Inspecting and controlling runs

The API isn’t only for starting things. A caller that triggered a run usually wants to poll it, and an ops script wants to list and pause. All read-only and idempotent, all the same bearer token:

# List recent runs of a DAG, newest first
curl -s -G "http://localhost:8080/api/v2/dags/bookshop_reprocess/dagRuns" \
  -H "Authorization: Bearer $TOKEN" \
  --data-urlencode "order_by=-logical_date" \
  --data-urlencode "limit=5"

# Fetch one run's state
curl -s "http://localhost:8080/api/v2/dags/bookshop_reprocess/dagRuns/manual__2026-06-24T00:00:00+00:00" \
  -H "Authorization: Bearer $TOKEN"

# Drill into the task instances of that run
curl -s "http://localhost:8080/api/v2/dags/bookshop_reprocess/dagRuns/manual__2026-06-24T00:00:00+00:00/taskInstances" \
  -H "Authorization: Bearer $TOKEN"

# Get the state of one specific task instance by task_id
curl -s "http://localhost:8080/api/v2/dags/bookshop_reprocess/dagRuns/manual__2026-06-24T00:00:00+00:00/taskInstances/reprocess" \
  -H "Authorization: Bearer $TOKEN"

That last call returns a single task instance — its state, try_number, start_date/end_date, and the hostname that ran it. A monitoring client that already knows which task it cares about (say the final publish step) polls that endpoint directly instead of listing all task instances and filtering client-side; it’s the difference between “is the whole run done?” and “did the one step I’m gating on finish?” A polling client loops on the run endpoint until state is success or failed, then reads taskInstances to find which task failed if it did. That’s the whole external-monitoring pattern in a handful of endpoints.

Pausing a DAG is a PATCH on the DAG resource — the API equivalent of flicking the toggle in the UI, and unpausing is the same call with the field flipped back:

# Pause: stop the scheduler from creating new runs
curl -s -X PATCH \
  "http://localhost:8080/api/v2/dags/bookshop_reprocess" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"is_paused": true}'

# Unpause: hand scheduling back once the deploy is done
curl -s -X PATCH \
  "http://localhost:8080/api/v2/dags/bookshop_reprocess" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"is_paused": false}'

This is how a deploy pipeline quiesces a DAG before swapping code under it: pause, wait for in-flight runs to drain, deploy, unpause. Pausing only stops new scheduled runs — it does not kill runs already in flight, which is exactly why the “wait to drain” step exists. The is_paused field is what the UI toggle writes, so scripting it and clicking it are the same operation.

One habit worth building: treat these tokens like production credentials. A JWT that can POST a run to your marts DAG can also trigger a full refresh at 2am on quarter close. Scope the identity, keep tokens short-lived, and don’t paste them into shared shell history. The API is powerful precisely because it’s the same surface the UI uses — which means anything a logged-in operator could do, a leaked token can do too.

TriggerDagRunOperator: one DAG starts another

Inside Airflow, the equivalent of that curl is the TriggerDagRunOperator. It lives in the standard provider — from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator — and it exists for one honest use case: a deliberate, named handoff between two DAGs that have a real boundary between them.

The bookshop’s staging pipeline and its marts pipeline are owned by different people and run on different cadences, but staging finishing should wake marts. Rather than merge them into one giant DAG, staging ends with a trigger:

from airflow.sdk import dag, task
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator


@dag(schedule="@daily")
def bookshop_staging():
    @task
    def stage_orders() -> str:
        BookshopWarehouseHook().refresh_staging()
        return "{{ ds }}"

    staged = stage_orders()

    hand_off = TriggerDagRunOperator(
        task_id="trigger_marts",
        trigger_dag_id="bookshop_marts",
        conf={"staged_date": "{{ ds }}", "source": "staging"},
        wait_for_completion=False,
        reset_dag_run=True,
    )

    staged >> hand_off


bookshop_staging()

The conf here is the same JSON payload the REST API sends — the marts DAG reads staged_date and source out of dag_run.conf exactly as if a curl had started it. The operator templates its conf, so {{ ds }} renders to the run date at execution time.

Four keyword arguments carry the real behavior, and knowing them is the difference between a clean handoff and a mystery:

  • wait_for_completion — when True, the trigger task doesn’t finish until the triggered run does. This turns a fire-and-forget into a synchronous dependency: your staging DAG’s trigger_marts task stays running (and can therefore fail) based on the marts run’s outcome. Leave it False for a true handoff; set it True when staging genuinely shouldn’t be considered done until marts succeeds.
  • allowed_states — paired with wait_for_completion, this says which final states count as success. Default is ["success"]; if you’d accept a marts run that ended success or skipped, list both. Anything outside the set fails the trigger task.
  • reset_dag_run — if a run already exists for the target logical_date, triggering again normally errors with a duplicate-run conflict. reset_dag_run=True clears and re-runs it instead. This is what makes the operator safe to retry: without it, a retried trigger task hits “run already exists” and fails for the wrong reason. With it, a retry cleanly re-triggers.
  • failed_states — the mirror image of allowed_states: the set of final states that should fail the trigger task immediately rather than keep it waiting. By default a target run that ends failed trips it, but you can widen the net — treat a marts run that ended skipped as a failed handoff, for instance, if “skipped” means the data you were waiting on never landed. Between allowed_states and failed_states you draw the exact line between “this counts as upstream succeeding,” “this means give up now,” and (anything in neither set) “keep polling.”
  • poke_interval — how often, in seconds, a completion-waiting trigger re-checks the target run’s state. It’s purely the polling cadence for wait_for_completion and does nothing without it. In classic (non-deferred) waiting it’s how long the task sleeps between checks while holding its slot; deferred, it’s how often the triggerer re-evaluates on the task’s behalf. A tight interval (10s) makes a short marts run feel responsive; a loose one (300s) is kinder to a long run you don’t need second-by-second news about. The poke_interval=60 in the example below says “check once a minute.”
  • deferrable — when waiting for completion, deferrable=True frees the worker slot while it waits. Instead of a task holding a slot for the entire marts run, the trigger task defers to the triggerer (the same mechanism the deferrable-sensors chapter covered) and wakes only when the target run finishes. On a busy cluster this is the difference between a handful of idle-waiting slots and none — and it’s why a synchronous handoff on a large fleet should almost always set deferrable=True rather than burn a slot polling.

Put together, a synchronous, retry-safe, slot-friendly handoff reads:

hand_off = TriggerDagRunOperator(
    task_id="trigger_marts",
    trigger_dag_id="bookshop_marts",
    conf={"staged_date": "{{ ds }}"},
    wait_for_completion=True,
    allowed_states=["success"],
    reset_dag_run=True,
    deferrable=True,
    poke_interval=60,
)

When Assets are the better answer

Before you reach for TriggerDagRunOperator, ask whether the relationship is really “DAG A commands DAG B” or actually “DAG B depends on data DAG A produces.” If it’s the latter — and cross-team handoffs usually are — the Asset-based approach from the Assets chapter is the cleaner contract. Staging produces an Asset; marts is scheduled on that Asset:

from airflow.sdk import Asset, dag, task

orders_staged = Asset("warehouse://bookshop/stg_orders")


@dag(schedule="@daily")
def bookshop_staging():
    @task(outlets=[orders_staged])
    def stage_orders():
        BookshopWarehouseHook().refresh_staging()

    stage_orders()


@dag(schedule=[orders_staged])   # runs when the Asset updates
def bookshop_marts():
    ...

The difference is who holds the knowledge. With TriggerDagRunOperator, the staging DAG has to know the name bookshop_marts and reach across the boundary to poke it — staging is coupled to its consumers. With Assets, staging only declares “I updated stg_orders” and any number of downstream DAGs subscribe without staging knowing they exist. Use TriggerDagRunOperator when you genuinely need to pass a conf payload or want the imperative, synchronous coupling (deploy gates, “recompute exactly this”); use Assets when the trigger is really about data readiness and you’d rather not hardcode who consumes it.

Params: the contract for a triggered run

A DAG that accepts conf from outside is accepting arbitrary JSON from a caller you may not control. A raw dict is not a schema, and “the run silently did the wrong thing because someone passed region: "usa" instead of "us"” is a genuinely bad afternoon. Airflow’s answer is the Param system: declare, per DAG, exactly which inputs you accept, their types, and their allowed values. Params are the typed contract; conf is what a caller sends to fill it.

You declare params on the DAG, and Airflow validates incoming conf against them before the run starts:

from airflow.sdk import dag, task
from airflow.sdk import Param


@dag(
    schedule=None,   # triggered, not scheduled
    params={
        "quarter": Param(
            "2026-Q2",
            type="string",
            pattern=r"^\d{4}-Q[1-4]$",
            description="Fiscal quarter to reprocess, e.g. 2026-Q2",
        ),
        "regions": Param(
            ["us", "eu"],
            type="array",
            items={"type": "string", "enum": ["us", "eu", "apac"]},
        ),
        "full_refresh": Param(False, type="boolean"),
        "batch_size": Param(500, type="integer", minimum=1, maximum=5000),
    },
)
def bookshop_reprocess():
    @task
    def reprocess(**context):
        conf = context["dag_run"].conf
        params = context["params"]
        quarter = params["quarter"]          # validated, with default applied
        regions = params["regions"]
        full = params["full_refresh"]
        BookshopWarehouseHook().reprocess(
            quarter=quarter, regions=regions, full_refresh=full
        )

    reprocess()


bookshop_reprocess()

Each Param carries a default (the first positional argument) plus JSON-Schema-style constraints: type, enum, minimum/maximum, pattern, and for arrays an items schema with its own enum. When a run is triggered — from the API, from TriggerDagRunOperator, or from the UI — Airflow merges the incoming conf over the defaults and validates the result. A conf with "quarter": "Q2" (no year) fails the pattern and the run is rejected up front, with an error naming the offending param, rather than blowing up three tasks deep with a confusing SQL error.

Inside tasks you have two ways to read the values, and they’re subtly different. context["params"] gives you the validated, defaulted view — every declared param is present, defaults filled in, types checked. context["dag_run"].conf gives you the raw payload the caller actually sent — useful if you want to know whether a value was explicitly provided versus defaulted. Reach for params for normal logic; reach for conf only when the distinction between “sent” and “defaulted” matters.

In templates, params render through the params namespace:

run_sql = SQLExecuteQueryOperator(
    task_id="reprocess_marts",
    conn_id="bookshop_dw",
    sql="CALL reprocess_orders('{{ params.quarter }}', {{ params.batch_size }})",
)

There’s a DagParam object underneath (what you get from dag.param("quarter") when wiring values into the TaskFlow graph directly), but for most work the params={...} declaration plus {{ params.x }} / context["params"] reads are all you touch.

The payoff shows up in the UI too. Because the params are typed and declared, the “Trigger DAG w/ config” form renders real inputs — a dropdown for the regions enum, a checkbox for the boolean full_refresh, a text box with pattern validation for quarter. An operator triggering a manual reprocess gets a form, not a “paste raw JSON and pray” box, and the same validation that guards the API guards the button. One declaration; three protected front doors.

Parse-time dynamic DAGs: many DAGs from one factory

The other meaning of “dynamic” is generating whole DAGs programmatically. This is a different mechanism from the dynamic task mapping of the earlier chapter, and confusing the two causes real pain. Task mapping fans out at runtime, inside one DAG — the count is decided when the run starts. DAG factories fan out at parse time, across many DAGs — the scheduler runs your factory code on every parse and registers whatever DAGs it produces.

The bookshop ingests from several source systems — the Shopify store, a wholesale partner feed, a returns portal — and each needs its own DAG so it can be paused, scheduled, and monitored independently. Rather than copy-paste three near-identical DAG files, describe them in YAML and generate them:

# dags/config/sources.yaml
sources:
  - id: shopify
    schedule: "@hourly"
    table: raw_shopify_orders
  - id: wholesale
    schedule: "@daily"
    table: raw_wholesale_orders
  - id: returns
    schedule: "0 */4 * * *"
    table: raw_returns

The factory reads that file and stamps out one DAG per entry, registering each into the module’s globals() so Airflow’s parser discovers it:

# dags/source_ingest_factory.py
from pathlib import Path

import yaml
from airflow.sdk import dag, task

CONFIG = Path(__file__).parent / "config" / "sources.yaml"


def build_dag(source: dict):
    dag_id = f"ingest_{source['id']}"

    @dag(dag_id=dag_id, schedule=source["schedule"], catchup=False, tags=["ingest"])
    def _factory():
        @task
        def extract() -> str:
            return SourceHook(source["id"]).dump_to_stage()

        @task
        def load(stage_path: str) -> int:
            return BookshopWarehouseHook().copy_into(
                table=source["table"], path=stage_path
            )

        load(extract())

    return _factory()


# Read config once, at parse time, and register one DAG per source.
with CONFIG.open() as fh:
    _sources = yaml.safe_load(fh)["sources"]

for _source in sorted(_sources, key=lambda s: s["id"]):
    _dag = build_dag(_source)
    globals()[f"ingest_{_source['id']}"] = _dag

The load-bearing line is globals()[f"ingest_{_source['id']}"] = _dag. Airflow’s DAG parser inspects the module’s top-level namespace for DAG objects, so a DAG that only exists inside a function is invisible — it has to land in globals() to be registered. Three YAML entries become three DAGs (ingest_shopify, ingest_wholesale, ingest_returns), each independently schedulable and pausable, from one file you maintain.

The parsing-cost tax

Here’s the tradeoff you must respect: that top-level code runs on every parse, and the scheduler parses your DAG files continuously — every dag_dir_list_interval, across every file, forever. Whatever the factory does at module top level, it pays for it on a loop, and slow parsing shows up as sluggish scheduling for the whole deployment, not just these DAGs.

So the rule is: parse-time code must be cheap and deterministic. Reading a local YAML file is fine. Sorting the list (so DAG order is stable across parses) is fine. What is not fine is the tempting version where the factory queries a database — for row in db.execute("SELECT * FROM tenants") at top level — or hits an API, or does heavy computation. That turns every parse into a database round-trip; multiply by the parse interval and the number of scheduler processes and you’ve built a self-inflicted load test against your metadata store’s neighbor. If the list of DAGs must come from a database, generate the YAML (or a JSON manifest) in a separate, scheduled job and let the factory read the cached file — keep the live query out of the parse path.

Sort your inputs, too. If the factory iterates a set or an unsorted dict, DAG discovery order can wobble between parses, which confuses versioning and makes diffs noisy. sorted(...) costs nothing and keeps the generated set stable.

When one mapped DAG beats many generated ones

Dynamic DAG generation and dynamic task mapping solve overlapping problems, and choosing wrong is a common design mistake. Generate separate DAGs when the units need separate lifecycles: their own schedule, their own pause switch, their own SLA and alerting, their own place in the UI. The three source systems qualify — wholesale runs daily and returns every four hours, and an operator wants to pause ingest_returns alone during a portal outage without touching Shopify.

Prefer one DAG with mapped tasks when the units share a lifecycle and differ only in data. The regional order files from the mapping chapter are the archetype: same schedule, same logic, same “process today’s batch,” varying only in which files showed up. Generating a DAG-per-region there would be pure overhead — dozens of near-identical DAGs cluttering the UI, all on the same schedule, when a single .expand() does the fan-out at runtime with none of the parse-time cost. If the only thing that varies is the data and everything runs on the same clock, that’s mapping’s job, not the factory’s.

Final thoughts

These four features are really one idea seen from four angles: the boundary between Airflow and the things that want to drive it. The REST API v2 is that boundary for outside systems, and Airflow 3.x hardened it — one stable, JWT-authed, versioned surface, with the experimental API retired. TriggerDagRunOperator is the boundary between two DAGs, and its keyword arguments (wait_for_completion, reset_dag_run, allowed_states, deferrable) let you choose exactly how tight that coupling is — while Assets stand ready when the handoff is really about data, not commands. Param is the boundary’s contract: it turns “arbitrary JSON from a caller” into a typed, validated, UI-rendered schema, so the API, the operator, and the trigger button all get the same guardrails from one declaration. And DAG factories are the boundary between your config and your DAG set — powerful, but taxed on every parse, so keep them cheap, local, and deterministic, and reach for mapping instead when the units only differ by data. Get these seams right and Airflow stops being the thing that only runs on a timer and becomes the thing the rest of your platform can safely call.

Examples in this chapter are docs-checked against the series’ stated versions, but they were not executed in this repository unless a companion project explicitly says so.

Next: Plugins, Listeners, and Notifiers

Comments