Synchronous Dag execution

Synchronous Dag execution refers to the ability in Airflow 3.1+ to trigger a Dag run using an API call and wait for it to complete before returning XCom values pushed by one or more tasks in the Dag run. This is useful both for single DAG runs and for cases where the same DAG may be triggered multiple times in parallel.

Synchronous Dag execution was added as an experimental feature in Airflow 3.1.

Assumed knowledge

Basic knowledge of Airflow. See Introduction to Apache Airflow.
Knowing how to use the Airflow REST API.
Basic understanding of XCom. See Passing data between tasks.

When to use synchronous Dag execution

Synchronous Dag execution is a way to use Airflow as the backend for services processing user requests coming from a frontend application like a website, mobile app, or slack bot. Common use cases include:

Inference execution: A user provides input to a pipeline that interacts with one or more LLMs and/or AI agents to generate a response. The response is served back to the user as soon as the Dag has completed running.
Ad-hoc requests: Non-technical stakeholders request data analyses that use a Dag to retrieve the desired result.
Data submission: Non-technical users can submit their data to a Dag to be processed and get immediate feedback on the status of the request and the result.

API endpoint

The endpoint to wait for a Dag run to complete is:

GET api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/wait

It includes the following path parameters:

dag_id: (Mandatory) The id of the DAG to wait for.
dag_run_id: (Mandatory) The id of the DAG run to wait for.

The query parameters are:

interval: (Mandatory) Seconds to wait between Dag run state checks.
result: (Optional) Array of strings or null. A list of task ids from which to pull the XCom value pushed under the return_value key. In Airflow 3.3+ you specify which task in a Dag returns the result in the Dag code, see specify the Dag result.

Calling this endpoint on any running Dag will start a waiting process until the Dag run completes. If any task is specified as returning a result or any XCom are requested in the result parameter, they are returned in the response upon Dag run completion.

Example script

The following script creates a Dag run for the my_dag Dag and waits for it to complete. It includes XComs pushed under the return_value key of the my_task task in the response.

import requests
from datetime import datetime
import json

_USERNAME = "admin"
_PASSWORD = "admin"
_HOST = "http://localhost:8080"  # To learn how to send API requests to Airflow running on Astro see: https://www.astronomer.io/docs/astro/airflow-api/

_DAG_ID = "my_dag"
_TASK_ID = "my_task"


def _get_jwt_token():
    token_url = f"{_HOST}/auth/token"
    payload = {"username": _USERNAME, "password": _PASSWORD}
    headers = {"Content-Type": "application/json"}
    response = requests.post(token_url, json=payload, headers=headers)

    token = response.json().get("access_token")
    return token


def _trigger_dag_run(dag_id: str):
    url = f"{_HOST}/api/v2/dags/{dag_id}/dagRuns"
    headers = {
        "Authorization": f"Bearer {_get_jwt_token()}",
        "Content-Type": "application/json",
    }
    payload = {
        "logical_date": None,
    }
    response = requests.post(url, headers=headers, json=payload)
    return response.json()["dag_run_id"]


def _wait_for_dag_run_completion(dag_id: str, dag_run_id: str):
    url = f"{_HOST}/api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/wait"
    headers = {
        "Authorization": f"Bearer {_get_jwt_token()}",
    }
    params = {
        "interval": 1,
        "result": [_TASK_ID],  # In Airflow 3.3+ the result task can also be defined in the Dag code
    }
    response = requests.get(url, headers=headers, params=params)
    print(f"Status Code: {response.status_code}")

    lines = response.text.strip().split("\n")
    json_objects = []

    for line in lines:
        if line.strip():
            json_obj = json.loads(line)
            json_objects.append(json_obj)
            print(f"Status: {json_obj.get('state', 'unknown')}")

    if json_objects:
        last_status_update = json_objects[-1]
        xcom_results = last_status_update.get("results", {})
        print("Last status update: ", last_status_update)
        print("XCom results: ", xcom_results)
        return xcom_results


if __name__ == "__main__":
    _dag_run_id = _trigger_dag_run(_DAG_ID)
    _wait_for_dag_run_completion(_DAG_ID, _dag_run_id)

Running the script above returns an output similar to the following:

Status Code: 200
Status: queued
Status: running
Status: running
Status: success
Last status update:  {'state': 'success', 'results': {'my_task': 'Hello World!'}}
XCom results:  {'my_task': 'Hello World!'}

Specify the Dag result

In Airflow 3.3+ you can define which task in a Dag returns the result the wait endpoint should return, using the @result decorator on top of a @task decorated function.

from airflow.sdk import task, result

@result
@task
def my_task():
    return "hello!"

my_task()

When using traditional operators, you can add a task’s .output (the XCom pushed with the key return_value) to the Dag object.

from airflow.sdk import DAG, chain
from airflow.providers.standard.operators.python import PythonOperator

def _my_task_func():
    return "hello!"

with DAG("my_dag") as dag:
    _my_task = PythonOperator(task_id="my_task", python_callable=_my_task_func)

    dag.add_result(_my_task.output)

Overview

Get started

Airflow concepts

Tutorials

Integrations & connections

Reference Architectures

Observability concepts

Glossary

Synchronous Dag execution

Assumed knowledge

When to use synchronous Dag execution

API endpoint

Example script

Specify the Dag result

​Assumed knowledge

​When to use synchronous Dag execution

​API endpoint

​Example script

​Specify the Dag result

Assumed knowledge

When to use synchronous Dag execution

API endpoint

Example script

Specify the Dag result