AIRFLOW-REST-API

Async wrapper for Airflow REST API. See more in documentation

INSTALL

pip install airflow-rest-api

USAGE

import asyncio
import os
import aiohttp
from airflow_rest_api.api import AirflowRestApi

AIRFLOW_HOST = os.environ.get("AIRFLOW_HOST", "http://127.0.0.1:8080")
AIRFLOW_USER = os.environ.get("AIRFLOW_USER", "admin")
AIRFLOW_PASSWORD = os.environ.get("AIRFLOW_PASSWORD", "admin")


async def main(ara):
    for template in ara.find_template("/dags"):
        template_id = template.id
        break
    auth = aiohttp.BasicAuth(AIRFLOW_USER, AIRFLOW_PASSWORD)
    async with aiohttp.ClientSession(auth=auth) as session:
        airflow_tasks = await ara.execute(session=session, template_id=template_id)
        if airflow_tasks.status == 200:
            file_token = airflow_tasks.raw["dags"][0]["file_token"]
            template_id = 360
            dag_source_code = (
                await ara.execute(
                    session=session,
                    method=ara.get_template(template_id).method,
                    url=ara.render_url(template_id=template_id, file_token=file_token)
                    )
                ).raw
            print(dag_source_code)


ara = AirflowRestApi(airflow_host=AIRFLOW_HOST)
ara.show_templates()

loop = asyncio.get_event_loop()
loop.run_until_complete(main(ara))