Airflow REST API - Data Management - Alibaba Cloud Documentation Center

REST API and CLI are two commonly used automation management tools in Airflow. The REST API allows you to automate the management of Airflow workflows (DAGs) and tasks through HTTP requests, while the CLI enables you to trigger and manage DAGs directly through commands.

For more information, see Airflow official documentation.

Using REST API

Obtain the credential (Token) required to log on to the Airflow instance.
Call the CreateAirflowLoginToken operation to obtain the token.
Note
The login token is valid for 2 hours. If you do not log on to the Airflow instance within 2 hours, you need to obtain a new token.

Log on to the Airflow instance.

Example of logging on to an instance using the cURL command:

curl -i "https://data-cn-beijing-dms.aliyuncs.com/airflow/33***/8691522017****/af-b3a7f110a6vmvn797****/login?token=xxxxx"

Note

Replace the following content in the request example: data-cn-beijing-dms.aliyuncs.com (region where the Airflow instance is located), airflow (instance name), 33*** (DMS tenant ID), 8691522017**** (workspace ID), af-b3a7797**** (Airflow instance ID), and Token.

Sample response:

HTTP/2 302
date: Thu, 05 Jun 2025 09:07:18 GMT
content-type: text/html; charset=utf-8
content-length: 309
vary: Origin
vary: Access-Control-Request-Method
vary: Access-Control-Request-Headers
server: envoy
location: /airflow/1/865084104****/af-ehrmszbxk735bl3x5****/home
cache-control: no-store
x-robots-tag: noindex, nofollow
x-envoy-upstream-service-time: 248
set-cookie: session=xxxxxxx; Expires=Sat, 05 Jul 2025 09:07:18 GMT; Secure; HttpOnly; Path=/; SameSite=None; Partitioned
x-content-type-options: nosniff
x-xss-protection: 0
referrer-policy: no-referrer
eagleeye-traceid: 0a032a1517491144384047056ec81f
strict-transport-security: max-age=31536000
timing-allow-origin: *

<!doctype html>
<html lang=en>
<title>Redirecting...</title>
<h1>Redirecting...</h1>
<p>You should be redirected automatically to the target URL: <a href="/airflow/1/8650841047****/af-ehrmszbxk735bl3x5v****/home">/airflow/1/8650841047****/af-ehrmszbxk735bl3x5v****/home</a>. If not, click the link.

After you log on to the instance, DMS automatically returns the session information. This session information is used to request the REST API.

Call the REST API.

You can refer to the Community Airflow REST API documentation to call the API.

Sample request:

curl 'https://data-cn-beijing-dms.aliyuncs.com/airflow/33***/8691522017****/af-b3a7f110a6vmvn797y****/api/v1/health' -b 'session=xxx'

Sample response:

{
  "dag_processor": {
    "latest_dag_processor_heartbeat": null,
    "status": null
  },
  "metadatabase": {
    "status": "healthy"
  },
  "scheduler": {
    "latest_scheduler_heartbeat": "2025-06-05T09:13:03.075907+00:00",
    "status": "healthy"
  },
  "triggerer": {
    "latest_triggerer_heartbeat": null,
    "status": null
  }
}

Using CLI command line

Obtain the credential (Token) required to log on to the Airflow instance.
Call the CreateAirflowLoginToken operation to obtain the token.
Note
The login token is valid for 2 hours. If you do not log on to the Airflow instance within 2 hours, you need to obtain a new token.

Log on to the Airflow instance.

After you log on to the instance, DMS automatically returns the session information. This session information is used to execute Airflow commands using CLI.

The following example shows how to send a request to log on to an instance using the cURL command:

curl -i "https://data-cn-beijing-dms.aliyuncs.com/airflow/33***/8691522017****/af-b3a7f110a6vmvn797****/login?token=xxxxx"

Note

Sample response:

HTTP/2 302
date: Thu, 05 Jun 2025 09:07:18 GMT
content-type: text/html; charset=utf-8
content-length: 309
vary: Origin
vary: Access-Control-Request-Method
vary: Access-Control-Request-Headers
server: envoy
location: /airflow/1/865084104****/af-ehrmszbxk735bl3x5****/home
cache-control: no-store
x-robots-tag: noindex, nofollow
x-envoy-upstream-service-time: 248
set-cookie: session=xxxxxxx; Expires=Sat, 05 Jul 2025 09:07:18 GMT; Secure; HttpOnly; Path=/; SameSite=None; Partitioned
x-content-type-options: nosniff
x-xss-protection: 0
referrer-policy: no-referrer
eagleeye-traceid: 0a032a1517491144384047056ec81f
strict-transport-security: max-age=31536000
timing-allow-origin: *

<!doctype html>
<html lang=en>
<title>Redirecting...</title>
<h1>Redirecting...</h1>
<p>You should be redirected automatically to the target URL: <a href="/airflow/1/8650841047****/af-ehrmszbxk735bl3x5v****/home">/airflow/1/8650841047****/af-ehrmszbxk735bl3x5v****/home</a>. If not, click the link.

Call the CLI interface to execute Airflow commands.

Sample request:

curl  -X 'POST' 'https://data-cn-beijing-dms.aliyuncs.com/airflow/33***/86915220175****/af-b3a7f110a6vmvn797y****/api/v1/command?command=version' -b 'session=faf9009b-exxx'

Sample response:

{
  "stderr": "",
  "stdout": "2.10.4\n"
}

Example: Calling REST API in Python environment

Execute the following command to install the DMS SDK.
```
pip install alibabacloud_dms20250414
```

Call the REST API.

Create the following Python script example, with the file name dms_rest_api.py.

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import os
import sys

from typing import List

from alibabacloud_dms20250414.client import Client as Dms20250414Client
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_dms20250414.models import CreateAirflowLoginTokenResponseBodyData
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dms20250414 import models as dms_20250414_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
import requests

endpoints = {
    "cn-beijing": "dms.cn-beijing.aliyuncs.com",
    "cn-hangzhou": "dms.cn-hangzhou.aliyuncs.com",
    "cn-shanghai": "dms.cn-shanghai.aliyuncs.com",
    "cn-shenzhen": "dms.cn-shenzhen.aliyuncs.com",
    "ap-southeast-1": "dms.ap-southeast-1.aliyuncs.com"
}


class DmsAirflowRestApi:

    def __init__(self, endpoint: str):
        """
        # Initialize the account Client with credentials
        @return: Client
        @throws Exception
        """
        # For production code, we recommend using a more secure approach without AccessKey. For credential configuration methods, see: https://www.alibabacloud.com/help/document_detail/378659.html.
        credential = CredentialClient()
        config = open_api_models.Config(
            credential=credential
        )
        # For Endpoint, refer to https://api.aliyun.com/product/Dms
        config.endpoint = endpoint
        self.client = Dms20250414Client(config)

    def get_login_token(self, airflowId: str) -> CreateAirflowLoginTokenResponseBodyData:
        create_airflow_login_token_request = dms_20250414_models.CreateAirflowLoginTokenRequest(
            airflow_id=airflowId
        )
        runtime = util_models.RuntimeOptions()
        try:
            # Write your code to display the response of the API operation if necessary.
            response = self.client.create_airflow_login_token_with_options(create_airflow_login_token_request, runtime)
            data = response.body.data
            return data
        except Exception as error:
            # Handle exceptions with caution in your actual business scenario and never ignore exceptions in your project. In this example, error messages are printed.
            # Print an error message.
            print(error.message)
            # Display information for troubleshooting.
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    def get_session_cookie(self, login_token: CreateAirflowLoginTokenResponseBodyData):
        login_url = f'{login_token.host}/login?token={login_token.token}'
        try:
            # Send the request.
            response = requests.get(login_url)
            response.raise_for_status()

            print(response.headers)
            # Get the session cookie
            if 'session' in response.cookies:
                return response.cookies['session']

            return None

        except Exception as e:
            print(f"Error: {e}")
            return None

    def list_dags(self, login_token: CreateAirflowLoginTokenResponseBodyData, session_cookie: str):
        login_url = f'{login_token.host}/api/v1/dags'
        try:
            cookies = {'session': session_cookie}
            response = requests.get(login_url, cookies=cookies)
            response.raise_for_status()

            print(response.json())

        except Exception as e:
            print(f"Error: {e}")


if __name__ == '__main__':
    region = sys.argv[1]
    airflowId = sys.argv[2]

    endpoint = endpoints.get(region)
    restApi = DmsAirflowRestApi(endpoint)

    login_token = restApi.get_login_token(airflowId)
    session_cookie = restApi.get_session_cookie(login_token)
    restApi.list_dags(login_token, session_cookie)

Note

You can call the CreateAirflowLoginToken operation to obtain the AirflowID.

Configure AccessKey ID and AccessKey secret.

export ALIBABA_CLOUD_ACCESS_KEY_ID=xxx
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=xxx

Run the Python script.
```
python3 dms_rest_api.py ${region_id} ${airflow_id}
```
Example: python3 dms_rest_api.py cn-hangzhou af-sxsssxx.

Supported CLI commands

For information about how to use CLI commands, see Official CLI documentation.

cheat-sheet connections add
connections delete
dags backfill
dags delete
dags list
dags list-jobs
dags list-import-errors
dags list-runs
dags next-execution
dags pause
dags report
dags reserialize
dags show
dags state
dags test
dags trigger
dags unpause
db clean
providers behaviours
providers get
providers hooks
providers list
providers links
providers notifications
providers secrets
providers triggerer
providers widgets
tasks clear
tasks failed-deps
tasks list
tasks render
tasks state
tasks states-for-dag-run
tasks test
variables delete
variables get
variables set
variables list
version