DAG to test reschedule mode sensors with lots of reschedules

reschedule_hell_dag.py

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.sensors.external_task_sensor import ExternalTaskSensor

# Default Arguments for the DAG
default_args = {
    "owner": "me",
    "start_date": datetime(2022, 1, 1),
    "depends_on_past": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
    "email": ["me@example.com"],
    "email_on_failure": False,
    "email_on_retry": False,
    "catchup": False,
}

# Create the DAG
dag = DAG(
    "reschedule_hell_dag",
    default_args=default_args,
    is_paused_upon_creation=False,
    schedule_interval="0 * * * *",  # schedule to run every hour
)

# BashOperator that sleeps for 60 minutes
sleep_60m = BashOperator(task_id="sleep_60m", bash_command="sleep 3600", dag=dag)

# ExternalTaskSensor that waits for the sleep_60m task to complete
wait_for_sleep_60m = ExternalTaskSensor(
    task_id="wait_for_sleep_60m",
    external_dag_id="reschedule_hell_dag",
    external_task_id="sleep_60m",
    execution_delta=timedelta(seconds=0),
    poke_interval=2,  # check every 2 seconds
    mode="reschedule",
    dag=dag,
)