Airflow 2.0.0 Async HTTP Request with aiohttp DAG example

async.py

import aiohttp
import asyncio

from airflow.decorators import dag, task
from airflow.operators.python import task
from datetime import date, datetime, timedelta

async def async_main():
  async with aiohttp.ClientSession() as session:
    async with session.get('http://httpbin.org/get') as resp:
      print(resp.status)
      print(await resp.text())
  
default_args = {
  'owner': 'Airflow',
  'retries': 0
}

@dag('AsyncExample', default_args=default_args, schedule_interval='@once', start_date=datetime(2021, 1, 26), catchup=False)
def async_dag():
  @task()
  def main() -> None:
    try:
      loop = asyncio.get_event_loop()
      loop.run_until_complete(async_main())
    except Exception as e:
      print(e)
    finally:
      loop.close()
  
  main()
  
dag = async_dag()