apache · hussein-awala · Nov 2, 2023 · Nov 6, 2023 · Nov 6, 2023 · Nov 7, 2023
diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml
@@ -3572,7 +3572,9 @@ components:
             timezone:
               $ref: "#/components/schemas/Timezone"
             catchup:
-              type: boolean
+              oneOf:
+                - type: string
+                - type: boolean
               readOnly: true
             orientation:
               type: string

diff --git a/airflow/api_connexion/schemas/dag_schema.py b/airflow/api_connexion/schemas/dag_schema.py
@@ -95,7 +95,7 @@ class DAGDetailSchema(DAGSchema):
 
     owners = fields.Method("get_owners", dump_only=True)
     timezone = TimezoneField()
-    catchup = fields.Boolean()
+    catchup = fields.Method("get_catchup", dump_only=True)
     orientation = fields.String()
     concurrency = fields.Method("get_concurrency")  # TODO: Remove in Airflow 3.0
     max_active_tasks = fields.Integer()
@@ -148,6 +148,11 @@ def get_params(obj: DAG):
         params = obj.params
         return {k: v.dump() for k, v in params.items()}
 
+    @staticmethod
+    def get_catchup(obj: DAG):
+        """Get catchup value defined in a DAG."""
+        return obj._backport_catchup
+
 
 class DAGCollection(NamedTuple):
     """List of DAGs with metadata."""

diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
@@ -2182,16 +2182,17 @@ scheduler:
       default: "10.0"
     catchup_by_default:
       description: |
-        Turn off scheduler catchup by setting this to ``False``.
+        Turn off scheduler catchup by setting this to ``disable``.
         Default behavior is unchanged and
         Command Line Backfills still work, but the scheduler
-        will not do scheduler catchup if this is ``False``,
+        will not do scheduler catchup if this is ``disable``,
         however it can be set on a per DAG basis in the
-        DAG definition (catchup)
+        DAG definition (catchup). You can set it to ``ignore_first``
+        to disable backfilling since the start date of the DAG.
       version_added: ~
-      type: boolean
+      type: string
       example: ~
-      default: "True"
+      default: "enable"
     ignore_first_depends_on_past_by_default:
       description: |
         Setting this to True will make first task instance of a task

diff --git a/airflow/example_dags/example_bash_operator.py b/airflow/example_dags/example_bash_operator.py
@@ -30,7 +30,7 @@
     dag_id="example_bash_operator",
     schedule="0 0 * * *",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     dagrun_timeout=datetime.timedelta(minutes=60),
     tags=["example", "example2"],
     params={"example_key": "example_value"},

diff --git a/airflow/example_dags/example_branch_datetime_operator.py b/airflow/example_dags/example_branch_datetime_operator.py
@@ -30,7 +30,7 @@
 dag1 = DAG(
     dag_id="example_branch_datetime_operator",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
     schedule="@daily",
 )
@@ -56,7 +56,7 @@
 dag2 = DAG(
     dag_id="example_branch_datetime_operator_2",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
     schedule="@daily",
 )
@@ -81,7 +81,7 @@
 dag3 = DAG(
     dag_id="example_branch_datetime_operator_3",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
     schedule="@daily",
 )

diff --git a/airflow/example_dags/example_branch_day_of_week_operator.py b/airflow/example_dags/example_branch_day_of_week_operator.py
@@ -30,7 +30,7 @@
 with DAG(
     dag_id="example_weekday_branch_operator",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
     schedule="@daily",
 ) as dag:

diff --git a/airflow/example_dags/example_branch_labels.py b/airflow/example_dags/example_branch_labels.py
@@ -30,7 +30,7 @@
     "example_branch_labels",
     schedule="@daily",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
 ) as dag:
     ingest = EmptyOperator(task_id="ingest")
     analyse = EmptyOperator(task_id="analyze")

diff --git a/airflow/example_dags/example_branch_operator.py b/airflow/example_dags/example_branch_operator.py
@@ -46,7 +46,7 @@
 with DAG(
     dag_id="example_branch_operator",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     schedule="@daily",
     tags=["example", "example2"],
     orientation="TB",

diff --git a/airflow/example_dags/example_branch_operator_decorator.py b/airflow/example_dags/example_branch_operator_decorator.py
@@ -40,7 +40,7 @@
 with DAG(
     dag_id="example_branch_python_operator_decorator",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     schedule="@daily",
     tags=["example", "example2"],
     orientation="TB",

diff --git a/airflow/example_dags/example_branch_python_dop_operator_3.py b/airflow/example_dags/example_branch_python_dop_operator_3.py
@@ -49,7 +49,7 @@ def should_run(**kwargs) -> str:
     dag_id="example_branch_dop_operator_v3",
     schedule="*/1 * * * *",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     default_args={"depends_on_past": True},
     tags=["example"],
 ) as dag:

diff --git a/airflow/example_dags/example_complex.py b/airflow/example_dags/example_complex.py
@@ -30,7 +30,7 @@
     dag_id="example_complex",
     schedule=None,
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example", "example2", "example3"],
 ) as dag:
     # Create

diff --git a/airflow/example_dags/example_dag_decorator.py b/airflow/example_dags/example_dag_decorator.py
@@ -45,7 +45,7 @@ def execute(self, context: Context):
 @dag(
     schedule=None,
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 )
 def example_dag_decorator(email: str = "example@example.com"):

diff --git a/airflow/example_dags/example_datasets.py b/airflow/example_dags/example_datasets.py
@@ -51,7 +51,7 @@
 
 with DAG(
     dag_id="dataset_produces_1",
-    catchup=False,
+    catchup="disable",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
     schedule="@daily",
     tags=["produces", "dataset-scheduled"],
@@ -62,7 +62,7 @@
 
 with DAG(
     dag_id="dataset_produces_2",
-    catchup=False,
+    catchup="disable",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
     schedule=None,
     tags=["produces", "dataset-scheduled"],
@@ -72,7 +72,7 @@
 # [START dag_dep]
 with DAG(
     dag_id="dataset_consumes_1",
-    catchup=False,
+    catchup="disable",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
     schedule=[dag1_dataset],
     tags=["consumes", "dataset-scheduled"],
@@ -86,7 +86,7 @@
 
 with DAG(
     dag_id="dataset_consumes_1_and_2",
-    catchup=False,
+    catchup="disable",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
     schedule=[dag1_dataset, dag2_dataset],
     tags=["consumes", "dataset-scheduled"],
@@ -99,7 +99,7 @@
 
 with DAG(
     dag_id="dataset_consumes_1_never_scheduled",
-    catchup=False,
+    catchup="disable",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
     schedule=[
         dag1_dataset,
@@ -115,7 +115,7 @@
 
 with DAG(
     dag_id="dataset_consumes_unknown_never_scheduled",
-    catchup=False,
+    catchup="disable",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
     schedule=[
         Dataset("s3://unrelated/dataset3.txt"),

diff --git a/airflow/example_dags/example_dynamic_task_mapping_with_no_taskflow_operators.py b/airflow/example_dags/example_dynamic_task_mapping_with_no_taskflow_operators.py
@@ -53,7 +53,7 @@ def execute(self, context):
 with DAG(
     dag_id="example_dynamic_task_mapping_with_no_taskflow_operators",
     start_date=datetime(2022, 3, 4),
-    catchup=False,
+    catchup="disable",
 ):
     # map the task to a list of values
     add_one_task = AddOneOperator.partial(task_id="add_one").expand(value=[1, 2, 3])

diff --git a/airflow/example_dags/example_external_task_marker_dag.py b/airflow/example_dags/example_external_task_marker_dag.py
@@ -50,7 +50,7 @@
 with DAG(
     dag_id="example_external_task_marker_parent",
     start_date=start_date,
-    catchup=False,
+    catchup="disable",
     schedule=None,
     tags=["example2"],
 ) as parent_dag:
@@ -66,7 +66,7 @@
     dag_id="example_external_task_marker_child",
     start_date=start_date,
     schedule=None,
-    catchup=False,
+    catchup="disable",
     tags=["example2"],
 ) as child_dag:
     # [START howto_operator_external_task_sensor]

diff --git a/airflow/example_dags/example_kubernetes_executor.py b/airflow/example_dags/example_kubernetes_executor.py
@@ -47,7 +47,7 @@
         dag_id="example_kubernetes_executor",
         schedule=None,
         start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-        catchup=False,
+        catchup="disable",
         tags=["example3"],
     ) as dag:
         # You can use annotations on your kubernetes pods!

diff --git a/airflow/example_dags/example_latest_only.py b/airflow/example_dags/example_latest_only.py
@@ -28,7 +28,7 @@
     dag_id="latest_only",
     schedule=datetime.timedelta(hours=4),
     start_date=datetime.datetime(2021, 1, 1),
-    catchup=False,
+    catchup="disable",
     tags=["example2", "example3"],
 ) as dag:
     latest_only = LatestOnlyOperator(task_id="latest_only")

diff --git a/airflow/example_dags/example_latest_only_with_trigger.py b/airflow/example_dags/example_latest_only_with_trigger.py
@@ -34,7 +34,7 @@
     dag_id="latest_only_with_trigger",
     schedule=datetime.timedelta(hours=4),
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example3"],
 ) as dag:
     latest_only = LatestOnlyOperator(task_id="latest_only")

diff --git a/airflow/example_dags/example_local_kubernetes_executor.py b/airflow/example_dags/example_local_kubernetes_executor.py
@@ -45,7 +45,7 @@
         dag_id="example_local_kubernetes_executor",
         schedule=None,
         start_date=datetime(2021, 1, 1),
-        catchup=False,
+        catchup="disable",
         tags=["example3"],
     ) as dag:
         # You can use annotations on your kubernetes pods!

diff --git a/airflow/example_dags/example_nested_branch_dag.py b/airflow/example_dags/example_nested_branch_dag.py
@@ -32,7 +32,7 @@
 with DAG(
     dag_id="example_nested_branch_dag",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     schedule="@daily",
     tags=["example"],
 ) as dag:

diff --git a/airflow/example_dags/example_params_trigger_ui.py b/airflow/example_dags/example_params_trigger_ui.py
@@ -40,7 +40,7 @@
     doc_md=__doc__,
     schedule=None,
     start_date=datetime.datetime(2022, 3, 4),
-    catchup=False,
+    catchup="disable",
     tags=["example_ui"],
     params={
         "names": Param(

diff --git a/airflow/example_dags/example_params_ui_tutorial.py b/airflow/example_dags/example_params_ui_tutorial.py
@@ -36,7 +36,7 @@
     doc_md=__doc__,
     schedule=None,
     start_date=datetime.datetime(2022, 3, 4),
-    catchup=False,
+    catchup="disable",
     tags=["example_ui"],
     params={
         # Let's start simple: Standard dict values are detected from type and offered as entry form fields.

diff --git a/airflow/example_dags/example_passing_params_via_test_command.py b/airflow/example_dags/example_passing_params_via_test_command.py
@@ -61,7 +61,7 @@ def print_env_vars(test_mode=None):
     "example_passing_params_via_test_command",
     schedule="*/1 * * * *",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     dagrun_timeout=datetime.timedelta(minutes=4),
     tags=["example"],
 ) as dag:

diff --git a/airflow/example_dags/example_python_operator.py b/airflow/example_dags/example_python_operator.py
@@ -45,7 +45,7 @@ def x():
     dag_id="example_python_operator",
     schedule=None,
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 ) as dag:
     # [START howto_operator_python]

diff --git a/airflow/example_dags/example_sensor_decorator.py b/airflow/example_dags/example_sensor_decorator.py
@@ -34,7 +34,7 @@
 @dag(
     schedule=None,
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 )
 def example_sensor_decorator():

diff --git a/airflow/example_dags/example_sensors.py b/airflow/example_dags/example_sensors.py
@@ -49,7 +49,7 @@ def failure_callable():
     dag_id="example_sensors",
     schedule=None,
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 ) as dag:
     # [START example_time_delta_sensor]

diff --git a/airflow/example_dags/example_setup_teardown.py b/airflow/example_dags/example_setup_teardown.py
@@ -27,7 +27,7 @@
 with DAG(
     dag_id="example_setup_teardown",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 ) as dag:
     root_setup = BashOperator(task_id="root_setup", bash_command="echo 'Hello from root_setup'").as_setup()

diff --git a/airflow/example_dags/example_setup_teardown_taskflow.py b/airflow/example_dags/example_setup_teardown_taskflow.py
@@ -26,7 +26,7 @@
 with DAG(
     dag_id="example_setup_teardown_taskflow",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 ) as dag:
 

diff --git a/airflow/example_dags/example_short_circuit_decorator.py b/airflow/example_dags/example_short_circuit_decorator.py
@@ -25,7 +25,7 @@
 from airflow.utils.trigger_rule import TriggerRule
 
 
-@dag(start_date=pendulum.datetime(2021, 1, 1, tz="UTC"), catchup=False, tags=["example"])
+@dag(start_date=pendulum.datetime(2021, 1, 1, tz="UTC"), catchup="disable", tags=["example"])
 def example_short_circuit_decorator():
     # [START howto_operator_short_circuit]
     @task.short_circuit()

diff --git a/airflow/example_dags/example_short_circuit_operator.py b/airflow/example_dags/example_short_circuit_operator.py
@@ -29,7 +29,7 @@
 with DAG(
     dag_id="example_short_circuit_operator",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 ) as dag:
     cond_true = ShortCircuitOperator(

diff --git a/airflow/example_dags/example_skip_dag.py b/airflow/example_dags/example_skip_dag.py
@@ -63,7 +63,7 @@ def create_test_pipeline(suffix, trigger_rule):
 with DAG(
     dag_id="example_skip_dag",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 ) as dag:
     create_test_pipeline("1", TriggerRule.ALL_SUCCESS)

diff --git a/airflow/example_dags/example_sla_dag.py b/airflow/example_dags/example_sla_dag.py
@@ -42,7 +42,7 @@ def sla_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
 @dag(
     schedule="*/2 * * * *",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     sla_miss_callback=sla_callback,
     default_args={"email": "email@example.com"},
 )

diff --git a/airflow/example_dags/example_task_group.py b/airflow/example_dags/example_task_group.py
@@ -29,7 +29,7 @@
 with DAG(
     dag_id="example_task_group",
     start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
-    catchup=False,
+    catchup="disable",
     tags=["example"],
 ) as dag:
     start = EmptyOperator(task_id="start")