中文版文档

Airflow Extended API Plugin

Airflow Extended API, which export airflow CLI command as REST-ful API to extend the ability of airflow official API.

This plugin is available for airflow 2.x Version and extensible, as you can easily define your own API to execute any Airflow CLI command so that it fits your demand.

Current Supported Commands

The following commands are supported now, and more is coming.

• airflow dags backfill
• airflow tasks run
• airflow tasks clear

Plugin Install

1. Install the plugin via pip

  pip install airflow-extended-api

2. Restart the Airflow Web Server

3. Open Airflow UI in Docs - Extended API OpenAPI or http://localhost:8080/ to view extended API details in swagger UI.

Usage

Examples

curl request example:

curl -X POST --user "airflow:airflow" https://localhost:8080/api/extended/clear -H "Content-Type: application/json" -d '{"dagName": "string","downstream": true,"endDate": "2019-08-24T14:15:22Z","jobName": "string","startDate": "2019-08-24T14:15:22Z","username": "Extended API"}'

Response Schema:

{
  "executed_command": "string",
  "exit_code": 0,
  "output_info": [
    "string"
  ],
  "error_info": [
    "string"
  ]
}

curl without Credentials data

Note that you will need to pass credentials' data in --user "{username}:{password}" format, or you will get an Unauthorized error.

curl -X POST http://127.0.0.1:8080/api/extended/clear -H "Content-Type: application/json" -d '{"dagName": "string","downstream": true,"endDate": "2019-08-24T14:15:22Z","jobName": "string","startDate": "2019-08-24T14:15:22Z","username": "Extended API"}'

response

{
  "detail": null,
  "status": 401,
  "title": "Unauthorized",
  "type": "https://airflow.apache.org/docs/apache-airflow/2.2.5/stable-rest-api-ref.html#section/Errors/Unauthenticated"
}

curl with wrong CLI Command

curl -X POST --user "airflow:airflow"  http://127.0.0.1:8080/api/extended/clear -H "Content-Type: application/json" -d '{"dagName": "string","downstream": true,"endDate": "2019-08-24T14:15:22Z","jobName": "string","startDate": "2019-08-24T14:15:22Z","username": "Extended API"}'

response

{
  "error_info": [
    "Traceback (most recent call last):",
    "  File \"/home/airflow/.local/bin/airflow\", line 8, in <module>",
    "    sys.exit(main())",
    "  File \"/home/airflow/.local/lib/python3.7/site-packages/airflow/__main__.py\", line 48, in main",
    "    args.func(args)",
    "  File \"/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/cli_parser.py\", line 48, in command",
    "    return func(*args, **kwargs)",
    "  File \"/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py\", line 92, in wrapper",
    "    return f(*args, **kwargs)",
    "  File \"/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/task_command.py\", line 506, in task_clear",
    "    dags = get_dags(args.subdir, args.dag_id, use_regex=args.dag_regex)",
    "  File \"/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py\", line 203, in get_dags",
    "    return [get_dag(subdir, dag_id)]",
    "  File \"/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py\", line 193, in get_dag",
    "    f\"Dag {dag_id!r} could not be found; either it does not exist or it failed to parse.\"",
    "airflow.exceptions.AirflowException: Dag 'string' could not be found; either it does not exist or it failed to parse.",
    ""
  ],
  "executed_command": "airflow tasks clear string -e 2019-08-24T14:15:22+00:00 -s 2019-08-24T14:15:22+00:00 -t string -y -d",
  "exit_code": 1,
  "output_info": [
    "[\u001b[34m2022-04-22 10:05:50,538\u001b[0m] {\u001b[34mdagbag.py:\u001b[0m500} INFO\u001b[0m - Filling up the DagBag from /opt/airflow/dags\u001b[0m",
    ""
  ]
}

Project Plan

• support custom configuration

Links and References

• Airflow configuration documentation
• Airflow CLI command documentation
• This project was inspired by the following projects:
  • andreax79/airflow-code-editor
  • airflow-plugins/airflow_api_plugin
• Contact email: Eric Cao itsericsmail@gmail.com