Unify DAG schedule args and change default to None

The arguments 'schedule_interval' and 'timetable' are removed from both
the DAG class and the `@dag` decorator.

The default value of the 'schedule' argument (on both entities) is
changed to None (i.e. a DAG will not have a schedule by default).

The 'timetable' attribute still exists on DAG, and is now the only value
that reflects the DAG's schedule. The 'schedule_interval' attribute is
removed from DAG.

The 'schedule_interval' on DagModel used to store a string
representation of DAG's attribute of the same name, is now replaced by
timetable_summary, which should (mostly?) work the same as before. We
can fix minor UI differences as we go. Some use cases that rely on that
field are also changed to use other fields instead (dataset_expression,
for example, can be used to check whether a DAG is dataset-triggered).

The API field 'schedule_interval' has also been removed since that field
no longer exists in the database. This has some side effects. The field
previously contains some type information for delta values, but now
becomes a simple string. It is unclear if anyone really needs the
information, but we can always bring it back if needed.

airflow/api_connexion/openapi/v1.yaml

@@ -3196,14 +3196,20 @@ components:
           description: >
             User-provided DAG description, which can consist of several sentences or paragraphs that
             describe DAG contents.
-        schedule_interval:
-          $ref: "#/components/schemas/ScheduleInterval"
+        timetable_summary:
+          type: string
+          readOnly: true
+          nullable: true
+          description: |
+            Timetable summary.
+
+            *New in version 3.0.0*
         timetable_description:
           type: string
           readOnly: true
           nullable: true
           description: |
-            Timetable/Schedule Interval description.
+            Timetable description.
 
             *New in version 2.3.0*
         tags:
@@ -5190,19 +5196,6 @@ components:
             *New in version 3.0.0*
 
     # Common data type
-    ScheduleInterval:
-      description: |
-        Schedule interval. Defines how often DAG runs, this object gets added to your latest task instance's
-        execution_date to figure out the next schedule.
-      nullable: true
-      readOnly: true
-      anyOf:
-        - $ref: "#/components/schemas/TimeDelta"
-        - $ref: "#/components/schemas/RelativeDelta"
-        - $ref: "#/components/schemas/CronExpression"
-      discriminator:
-        propertyName: __type
-
     TimeDelta:
       description: Time delta
       type: object

airflow/api_connexion/schemas/common_schema.py

@@ -24,7 +24,6 @@
 import marshmallow
 from dateutil import relativedelta
 from marshmallow import Schema, fields, validate
-from marshmallow_oneofschema import OneOfSchema
 
 from airflow.models.mappedoperator import MappedOperator
 from airflow.serialization.serialized_objects import SerializedBaseOperator
@@ -90,42 +89,6 @@ def make_cron_expression(self, data, **kwargs):
         return CronExpression(data["value"])
 
 
-class ScheduleIntervalSchema(OneOfSchema):
-    """
-    Schedule interval.
-
-    It supports the following types:
-
-    * TimeDelta
-    * RelativeDelta
-    * CronExpression
-    """
-
-    type_field = "__type"
-    type_schemas = {
-        "TimeDelta": TimeDeltaSchema,
-        "RelativeDelta": RelativeDeltaSchema,
-        "CronExpression": CronExpressionSchema,
-    }
-
-    def _dump(self, obj, update_fields=True, **kwargs):
-        if isinstance(obj, str):
-            obj = CronExpression(obj)
-
-        return super()._dump(obj, update_fields=update_fields, **kwargs)
-
-    def get_obj_type(self, obj):
-        """Select schema based on object type."""
-        if isinstance(obj, datetime.timedelta):
-            return "TimeDelta"
-        elif isinstance(obj, relativedelta.relativedelta):
-            return "RelativeDelta"
-        elif isinstance(obj, CronExpression):
-            return "CronExpression"
-        else:
-            raise TypeError(f"Unknown object type: {obj.__class__.__name__}")
-
-
 class ColorField(fields.String):
     """Schema for color property."""
 

airflow/api_connexion/schemas/dag_schema.py

@@ -22,7 +22,7 @@
 from marshmallow import Schema, fields
 from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
 
-from airflow.api_connexion.schemas.common_schema import ScheduleIntervalSchema, TimeDeltaSchema, TimezoneField
+from airflow.api_connexion.schemas.common_schema import TimeDeltaSchema, TimezoneField
 from airflow.configuration import conf
 from airflow.models.dag import DagModel, DagTag
 
@@ -63,7 +63,7 @@ class Meta:
     file_token = fields.Method("get_token", dump_only=True)
     owners = fields.Method("get_owners", dump_only=True)
     description = auto_field(dump_only=True)
-    schedule_interval = fields.Nested(ScheduleIntervalSchema)
+    timetable_summary = auto_field(dump_only=True)
     timetable_description = auto_field(dump_only=True)
     tags = fields.List(fields.Nested(DagTagSchema), dump_only=True)
     max_active_tasks = auto_field(dump_only=True)

airflow/cli/commands/dag_command.py

@@ -345,7 +345,7 @@ def _get_dagbag_dag_details(dag: DAG) -> dict:
         "file_token": None,
         "owners": dag.owner,
         "description": dag.description,
-        "schedule_interval": dag.schedule_interval,
+        "timetable_summary": dag.timetable.summary,
         "timetable_description": dag.timetable.description,
         "tags": dag.tags,
         "max_active_tasks": dag.max_active_tasks,

airflow/config_templates/config.yml

@@ -2521,7 +2521,7 @@ scheduler:
     allow_trigger_in_future:
       description: |
         Allow externally triggered DagRuns for Execution Dates in the future
-        Only has effect if schedule_interval is set to None in DAG
+        Only has effect if schedule is set to None in DAG
       version_added: 1.10.8
       type: boolean
       example: ~

airflow/exceptions.py

@@ -199,10 +199,6 @@ def __str__(self) -> str:
         return f"Ignoring DAG {self.dag_id} from {self.incoming} - also found in {self.existing}"
 
 
-class AirflowDagInconsistent(AirflowException):
-    """Raise when a DAG has inconsistent attributes."""
-
-
 class AirflowClusterPolicyViolation(AirflowException):
     """Raise when there is a violation of a Cluster Policy in DAG definition."""
 

airflow/migrations/versions/0004_3_0_0_rename_schedule_interval_to_timetable_.py

@@ -0,0 +1,60 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Rename DagModel schedule_interval to timetable_summary.
+
+Revision ID: 0bfc26bc256e
+Revises: d0f1c55954fa
+Create Date: 2024-08-15 06:24:14.363316
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0bfc26bc256e"
+down_revision = "d0f1c55954fa"
+branch_labels = None
+depends_on = None
+airflow_version = "3.0.0"
+
+
+def upgrade():
+    """Rename DagModel schedule_interval to timetable_summary."""
+    with op.batch_alter_table("dag", schema=None) as batch_op:
+        batch_op.alter_column(
+            "schedule_interval",
+            new_column_name="timetable_summary",
+            type_=sa.Text,
+            nullable=True,
+        )
+
+
+def downgrade():
+    """Rename timetable_summary back to schedule_interval."""
+    with op.batch_alter_table("dag", schema=None) as batch_op:
+        batch_op.alter_column(
+            "timetable_summary",
+            new_column_name="schedule_interval",
+            type_=sa.Text,
+            nullable=True,
+        )

airflow/models/baseoperator.py

@@ -50,7 +50,6 @@
 
 import attr
 import pendulum
-from dateutil.relativedelta import relativedelta
 from sqlalchemy import select
 from sqlalchemy.orm.exc import NoResultFound
 
@@ -120,8 +119,6 @@
     from airflow.utils.task_group import TaskGroup
     from airflow.utils.types import ArgNotSet
 
-ScheduleInterval = Union[str, timedelta, relativedelta]
-
 TaskPreExecuteHook = Callable[[Context], None]
 TaskPostExecuteHook = Callable[[Context, Any], None]
 
@@ -612,10 +609,10 @@ class derived from this one results in the creation of a task object,
     :param start_date: The ``start_date`` for the task, determines
         the ``execution_date`` for the first task instance. The best practice
         is to have the start_date rounded
-        to your DAG's ``schedule_interval``. Daily jobs have their start_date
+        to your DAG's schedule. Daily jobs have their start_date
         some day at 00:00:00, hourly jobs have their start_date at 00:00
         of a specific hour. Note that Airflow simply looks at the latest
-        ``execution_date`` and adds the ``schedule_interval`` to determine
+        ``execution_date`` and adds the schedule to determine
         the next ``execution_date``. It is also very important
         to note that different tasks' dependencies
         need to line up in time. If task A depends on task B and their