Adding ContinuousTimetable and support for @continuous schedule_interval

airflow/models/dag.py

@@ -87,7 +87,12 @@
 from airflow.stats import Stats
 from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction, Timetable
 from airflow.timetables.interval import CronDataIntervalTimetable, DeltaDataIntervalTimetable
-from airflow.timetables.simple import DatasetTriggeredTimetable, NullTimetable, OnceTimetable
+from airflow.timetables.simple import (
+    ContinuousTimetable,
+    DatasetTriggeredTimetable,
+    NullTimetable,
+    OnceTimetable,
+)
 from airflow.typing_compat import Literal
 from airflow.utils import timezone
 from airflow.utils.dag_cycle_tester import check_cycle
@@ -181,6 +186,8 @@ def create_timetable(interval: ScheduleIntervalArg, timezone: Timezone) -> Timet
         return NullTimetable()
     if interval == "@once":
         return OnceTimetable()
+    if interval == "@continuous":
+        return ContinuousTimetable()
     if isinstance(interval, (timedelta, relativedelta)):
         return DeltaDataIntervalTimetable(interval)
     if isinstance(interval, str):
@@ -546,6 +553,12 @@ def __init__(
         self.last_loaded = timezone.utcnow()
         self.safe_dag_id = dag_id.replace(".", "__dot__")
         self.max_active_runs = max_active_runs
+        if self.timetable.active_runs_limit is not None:
+            if self.timetable.active_runs_limit < self.max_active_runs:
+                raise AirflowException(
+                    f"Invalid max_active_runs: {type(self.timetable)} "
+                    f"requires max_active_runs <= {self.timetable.active_runs_limit}"
+                )
         self.dagrun_timeout = dagrun_timeout
         self.sla_miss_callback = sla_miss_callback
         if default_view in DEFAULT_VIEW_PRESETS:

airflow/timetables/base.py

@@ -135,6 +135,14 @@ class Timetable(Protocol):
     This should be a list of field names on the DAG run object.
     """
 
+    active_runs_limit: int | None = None
+    """Override the max_active_runs parameter of any DAGs using this timetable.
+    This is called during DAG initializing, and will set the max_active_runs if
+    it returns a value. In most cases this should return None, but in some cases
+    (for example, the ContinuousTimetable) there are good reasons for limiting
+    the DAGRun parallelism.
+    """
+
     @classmethod
     def deserialize(cls, data: dict[str, Any]) -> Timetable:
         """Deserialize a timetable from data.

airflow/timetables/simple.py

@@ -19,10 +19,11 @@
 import operator
 from typing import TYPE_CHECKING, Any, Collection
 
+from pendulum import DateTime
+
 from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction, Timetable
 
 if TYPE_CHECKING:
-    from pendulum import DateTime
     from sqlalchemy import Session
 
     from airflow.models.dataset import DatasetEvent
@@ -108,6 +109,41 @@ def next_dagrun_info(
         return DagRunInfo.exact(run_after)
 
 
+class ContinuousTimetable(_TrivialTimetable):
+    """Timetable that schedules continually, while still respecting start_date and end_date
+
+    This corresponds to ``schedule="@continuous"``.
+    """
+
+    description: str = "As frequently as possible, but only one run at a time."
+
+    active_runs_limit = 1  # Continuous DAGRuns should be constrained to one run at a time
+
+    @property
+    def summary(self) -> str:
+        return "@continuous"
+
+    def next_dagrun_info(
+        self,
+        *,
+        last_automated_data_interval: DataInterval | None,
+        restriction: TimeRestriction,
+    ) -> DagRunInfo | None:
+        if restriction.earliest is None:  # No start date, won't run.
+            return None
+        if last_automated_data_interval is not None:  # has already run once
+            start = last_automated_data_interval.end
+            end = DateTime.utcnow()
+        else:  # first run
+            start = restriction.earliest
+            end = max(restriction.earliest, DateTime.utcnow())  # won't run any earlier than start_date
+
+        if restriction.latest is not None and end > restriction.latest:
+            return None
+
+        return DagRunInfo.interval(start, end)
+
+
 class DatasetTriggeredTimetable(NullTimetable):
     """Timetable that never schedules anything.
 

docs/apache-airflow/core-concepts/dag-run.rst

@@ -57,6 +57,8 @@ or one of the following cron "presets". For more elaborate scheduling requiremen
 +----------------+--------------------------------------------------------------------+-----------------+
 | ``@once``      | Schedule once and only once                                        |                 |
 +----------------+--------------------------------------------------------------------+-----------------+
+| ``@continuous``| Run as soon as the previous run finishes                           |                 |
++----------------+--------------------------------------------------------------------+-----------------+
 | ``@hourly``    | Run once an hour at the end of the hour                            | ``0 * * * *``   |
 +----------------+--------------------------------------------------------------------+-----------------+
 | ``@daily``     | Run once a day at midnight (24:00)                                 | ``0 0 * * *``   |

tests/models/test_dag.py

@@ -63,7 +63,12 @@
 from airflow.security import permissions
 from airflow.templates import NativeEnvironment, SandboxedEnvironment
 from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction, Timetable
-from airflow.timetables.simple import DatasetTriggeredTimetable, NullTimetable, OnceTimetable
+from airflow.timetables.simple import (
+    ContinuousTimetable,
+    DatasetTriggeredTimetable,
+    NullTimetable,
+    OnceTimetable,
+)
 from airflow.utils import timezone
 from airflow.utils.file import list_py_file_paths
 from airflow.utils.session import create_session, provide_session
@@ -2407,6 +2412,21 @@ def test_schedule_dag_param(self, kwargs):
             with DAG(dag_id="hello", **kwargs):
                 pass
 
+    def test_continuous_schedule_interval_limits_max_active_runs(self):
+
+        dag = DAG("continuous", start_date=DEFAULT_DATE, schedule_interval="@continuous", max_active_runs=1)
+        assert isinstance(dag.timetable, ContinuousTimetable)
+        assert dag.max_active_runs == 1
+
+        dag = DAG("continuous", start_date=DEFAULT_DATE, schedule_interval="@continuous", max_active_runs=0)
+        assert isinstance(dag.timetable, ContinuousTimetable)
+        assert dag.max_active_runs == 0
+
+        with pytest.raises(AirflowException):
+            dag = DAG(
+                "continuous", start_date=DEFAULT_DATE, schedule_interval="@continuous", max_active_runs=25
+            )
+
 
 class TestDagModel:
     def _clean(self):

tests/timetables/test_continuous_timetable.py

@@ -0,0 +1,93 @@
+# 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.
+
+from __future__ import annotations
+
+import pendulum
+import pytest
+import time_machine
+
+from airflow.timetables.base import DataInterval, TimeRestriction
+from airflow.timetables.simple import ContinuousTimetable
+
+BEFORE_DATE = pendulum.datetime(2023, 3, 1, tz="UTC")
+START_DATE = pendulum.datetime(2023, 3, 3, tz="UTC")
+DURING_DATE = pendulum.datetime(2023, 3, 6, tz="UTC")
+END_DATE = pendulum.datetime(2023, 3, 10, tz="UTC")
+AFTER_DATE = pendulum.datetime(2023, 3, 12, tz="UTC")
+
+
+@pytest.fixture()
+def restriction():
+    return TimeRestriction(earliest=START_DATE, latest=END_DATE, catchup=True)
+
+
+@pytest.fixture()
+def timetable():
+    return ContinuousTimetable()
+
+
+def test_no_runs_without_start_date(timetable):
+    next_info = timetable.next_dagrun_info(
+        last_automated_data_interval=None,
+        restriction=TimeRestriction(earliest=None, latest=None, catchup=False),
+    )
+    assert next_info is None
+
+
+@time_machine.travel(DURING_DATE)
+def test_first_run_after_start_date_correct_interval(timetable, restriction):
+    next_info = timetable.next_dagrun_info(
+        last_automated_data_interval=None,
+        restriction=restriction,
+    )
+    assert next_info.run_after == DURING_DATE
+    assert next_info.data_interval.start == START_DATE
+    assert next_info.data_interval.end == DURING_DATE
+
+
+@time_machine.travel(BEFORE_DATE)
+def test_first_run_before_start_date_correct_interval(timetable, restriction):
+    next_info = timetable.next_dagrun_info(
+        last_automated_data_interval=None,
+        restriction=restriction,
+    )
+    assert next_info.run_after == START_DATE
+    assert next_info.data_interval.start == START_DATE
+    assert next_info.data_interval.end == START_DATE
+
+
+@time_machine.travel(DURING_DATE)
+def test_run_uses_utcnow(timetable, restriction):
+
+    next_info = timetable.next_dagrun_info(
+        last_automated_data_interval=DataInterval(START_DATE, DURING_DATE),
+        restriction=restriction,
+    )
+
+    assert next_info.run_after == DURING_DATE
+
+
+@time_machine.travel(AFTER_DATE)
+def test_no_runs_after_end_date(timetable, restriction):
+
+    next_info = timetable.next_dagrun_info(
+        last_automated_data_interval=DataInterval(START_DATE, DURING_DATE),
+        restriction=restriction,
+    )
+
+    assert next_info is None