Add AssetAndTimeSchedule timetable

airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst

@@ -147,6 +147,30 @@ If one asset is updated multiple times before all consumed assets update, the do
 
     }
 
+Gate scheduled runs on asset updates
+------------------------------------
+
+Use ``AssetAndTimeSchedule`` when you want a Dag to follow a normal time-based timetable but only start after specific assets have been updated. Airflow creates the DagRun at the scheduled time and keeps it queued until every required asset has queued an event. When the DagRun starts, those asset events are consumed so the next scheduled run waits for new updates. This does not create additional asset-triggered runs.
+
+If ``dagrun_timeout`` is set on the Dag, it also limits how long an ``AssetAndTimeSchedule`` run may remain queued while waiting for its asset condition to become ready.
+
+.. code-block:: python
+
+    from airflow.sdk import DAG, Asset, AssetAndTimeSchedule, CronTriggerTimetable
+
+    example_asset = Asset("s3://asset/example.csv")
+
+    with DAG(
+        dag_id="gated_hourly_dag",
+        schedule=AssetAndTimeSchedule(
+            timetable=CronTriggerTimetable("0 * * * *", timezone="UTC"),
+            assets=[example_asset],
+        ),
+        ...,
+    ):
+        ...
+
+
 Fetching information from a triggering asset event
 ----------------------------------------------------
 
@@ -416,6 +440,9 @@ Combining asset and time-based schedules
 
 AssetTimetable Integration
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-You can schedule Dags based on both asset events and time-based schedules using ``AssetOrTimeSchedule``. This allows you to create workflows when a Dag needs both to be triggered by data updates and run periodically according to a fixed timetable.
+Asset-aware timetables combine asset expressions with a time-based schedule:
+
+* Use ``AssetOrTimeSchedule`` to create runs both on a timetable and when assets update, producing scheduled runs and asset-triggered runs independently.
+* Use ``AssetAndTimeSchedule`` to keep a Dag on a timetable but only start those scheduled runs once the referenced assets have been updated.
 
-For more detailed information on ``AssetOrTimeSchedule``, refer to the corresponding section in :ref:`AssetOrTimeSchedule <asset-timetable-section>`.
+For more detailed information on asset-aware timetables, refer to :ref:`AssetOrTimeSchedule <asset-timetable-section>`.

airflow-core/docs/authoring-and-scheduling/timetable.rst

@@ -268,16 +268,18 @@ Asset event based scheduling with time based scheduling
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 Combining conditional asset expressions with time-based schedules enhances scheduling flexibility.
 
-The ``AssetOrTimeSchedule`` is a specialized timetable that allows for the scheduling of Dags based on both time-based schedules and asset events. It also facilitates the creation of both scheduled runs, as per traditional timetables, and asset-triggered runs, which operate independently.
+Asset-aware timetables let you combine a time-based schedule with an asset expression:
+
+* ``AssetOrTimeSchedule`` schedules Dag runs both on the timetable and whenever the assets update. It creates traditional scheduled runs and asset-triggered runs independently.
+* ``AssetAndTimeSchedule`` keeps the Dag on a time-based timetable but defers starting a queued run until all referenced assets are ready. When the run starts, the asset events are consumed so the next scheduled run waits for the next set of updates. No asset-triggered runs are created.
 
 This feature is particularly useful in scenarios where a Dag needs to run on asset updates and also at periodic intervals. It ensures that the workflow remains responsive to data changes and consistently runs regular checks or updates.
 
 Here's an example of a Dag using ``AssetOrTimeSchedule``:
 
 .. code-block:: python
 
-    from airflow.timetables.assets import AssetOrTimeSchedule
-    from airflow.timetables.trigger import CronTriggerTimetable
+    from airflow.sdk import AssetOrTimeSchedule, CronTriggerTimetable
 
 
     @dag(
@@ -290,6 +292,23 @@ Here's an example of a Dag using ``AssetOrTimeSchedule``:
         # Dag tasks go here
         pass
 
+Here's an example of a Dag using ``AssetAndTimeSchedule`` to require both the time-based schedule and fresh assets before a run starts:
+
+.. code-block:: python
+
+    from airflow.sdk import AssetAndTimeSchedule, CronTriggerTimetable
+
+
+    @dag(
+        schedule=AssetAndTimeSchedule(
+            timetable=CronTriggerTimetable("0 1 * * 3", timezone="UTC"),
+            assets=(dag1_asset & dag2_asset),
+        ),
+    )
+    def example_gated_dag():
+        # Dag tasks go here
+        pass
+
 
 
 Timetables comparisons

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -113,6 +113,7 @@
 from airflow.serialization.definitions.assets import SerializedAssetUniqueKey
 from airflow.serialization.definitions.notset import NOTSET
 from airflow.ti_deps.dependencies_states import ACTIVE_STATES, EXECUTION_STATES
+from airflow.timetables.assets import AssetAndTimeSchedule
 from airflow.timetables.base import compute_rollup_fingerprint
 from airflow.timetables.simple import AssetTriggeredTimetable, PartitionedAssetTimetable
 from airflow.utils.event_scheduler import EventScheduler
@@ -2542,6 +2543,7 @@ def _update_state(dag: SerializedDAG, dag_run: DagRun):
         cached_get_dag: Callable[[DagRun], SerializedDAG | None] = lru_cache()(
             partial(self.scheduler_dag_bag.get_dag_for_run, session=session)
         )
+        asset_evaluator = AssetEvaluator(session)
 
         for dag_run in dag_runs:
             dag_id = dag_run.dag_id
@@ -2582,6 +2584,55 @@ def _update_state(dag: SerializedDAG, dag_run: DagRun):
                         dag_run.run_id,
                     )
                     continue
+            # For AssetAndTimeSchedule, defer starting until all required assets are queued.
+            # Only gate scheduled runs; manual and backfill runs should start immediately.
+            if isinstance(dag.timetable, AssetAndTimeSchedule) and dag_run.run_type == DagRunType.SCHEDULED:
+                # Reuse dagrun_timeout to fail runs that wait in QUEUED for assets for too long.
+                if (
+                    dag.dagrun_timeout
+                    and dag_run.queued_at
+                    and dag_run.queued_at < timezone.utcnow() - dag.dagrun_timeout
+                ):
+                    dag_run.set_state(DagRunState.FAILED)
+                    session.flush()
+                    self.log.info(
+                        "Run %s of %s has timed-out while waiting for assets",
+                        dag_run.run_id,
+                        dag_run.dag_id,
+                    )
+                    if dag_run.dag_model is not None:
+                        self._set_exceeds_max_active_runs(dag_model=dag_run.dag_model, session=session)
+                    dag_run.notify_dagrun_state_changed(msg="timed_out")
+                    continue
+
+                queued_adrqs = session.scalars(
+                    with_row_locks(
+                        select(AssetDagRunQueue)
+                        .where(AssetDagRunQueue.target_dag_id == dag_id)
+                        .options(joinedload(AssetDagRunQueue.asset)),
+                        of=AssetDagRunQueue,
+                        session=session,
+                        skip_locked=True,
+                    )
+                ).all()
+                statuses = {
+                    SerializedAssetUniqueKey.from_asset(record.asset): True for record in queued_adrqs
+                }
+
+                if not asset_evaluator.run(dag.timetable.asset_condition, statuses=statuses):
+                    self.log.debug("Deferring DagRun until assets ready; dag_id=%s run_id=%s", dag_id, run_id)
+                    # Do not increment active run counts; we didn't start it.
+                    continue
+
+                if queued_adrqs:
+                    # Consume only the rows selected for this DagRun to avoid races with new asset events.
+                    adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
+                    session.execute(
+                        delete(AssetDagRunQueue).where(
+                            tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
+                        )
+                    )
+
             active_runs_of_dags[(dag_run.dag_id, backfill_id)] += 1
             _update_state(dag, dag_run)
             dag_run.notify_dagrun_state_changed(msg="started")

airflow-core/src/airflow/models/dag.py

@@ -751,6 +751,12 @@ def dag_ready(dag_id: str, cond: SerializedAssetBase, statuses: dict[UKey, bool]
         for ser_dag in ser_dags:
             dag_id = ser_dag.dag_id
             statuses = dag_statuses[dag_id]
+            timetable = ser_dag.dag.timetable
+
+            if not isinstance(timetable, AssetTriggeredTimetable):
+                del adrq_by_dag[dag_id]
+                continue
+
             ready = dag_ready(dag_id, cond=ser_dag.dag.timetable.asset_condition, statuses=statuses)
             if not ready:
                 log.debug("Asset condition not met for dag '%s'", dag_id)

airflow-core/src/airflow/serialization/encoders.py

@@ -33,6 +33,7 @@
     Asset,
     AssetAlias,
     AssetAll,
+    AssetAndTimeSchedule,
     AssetAny,
     AssetOrTimeSchedule,
     ChainMapper,
@@ -306,6 +307,7 @@ class _Serializer:
     """Serialization logic."""
 
     BUILTIN_TIMETABLES: dict[type, str] = {
+        AssetAndTimeSchedule: "airflow.timetables.assets.AssetAndTimeSchedule",
         AssetOrTimeSchedule: "airflow.timetables.assets.AssetOrTimeSchedule",
         AssetTriggeredTimetable: "airflow.timetables.simple.AssetTriggeredTimetable",
         ContinuousTimetable: "airflow.timetables.simple.ContinuousTimetable",
@@ -408,6 +410,13 @@ def _(self, timetable: MultipleCronTriggerTimetable) -> dict[str, Any]:
             "run_immediately": encode_run_immediately(representitive.run_immediately),
         }
 
+    @serialize_timetable.register
+    def _(self, timetable: AssetAndTimeSchedule) -> dict[str, Any]:
+        return {
+            "asset_condition": encode_asset_like(timetable.asset_condition),
+            "timetable": encode_timetable(timetable.timetable),
+        }
+
     @serialize_timetable.register
     def _(self, timetable: AssetOrTimeSchedule) -> dict[str, Any]:
         return {

airflow-core/src/airflow/timetables/assets.py

@@ -20,7 +20,10 @@
 import typing
 
 from airflow.exceptions import AirflowTimetableInvalid
-from airflow.serialization.definitions.assets import SerializedAsset, SerializedAssetBase
+from airflow.sdk.definitions.asset import BaseAsset  # TODO: Use serialized classes.
+from airflow.serialization.definitions.assets import SerializedAsset, SerializedAssetAll, SerializedAssetBase
+from airflow.serialization.encoders import ensure_serialized_asset
+from airflow.timetables.base import Timetable
 from airflow.timetables.simple import AssetTriggeredTimetable
 from airflow.utils.types import DagRunType
 
@@ -29,7 +32,7 @@
 
     import pendulum
 
-    from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction, Timetable
+    from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction
 
 
 class AssetOrTimeSchedule(AssetTriggeredTimetable):
@@ -90,3 +93,75 @@ def generate_run_id(self, *, run_type: DagRunType, **kwargs: typing.Any) -> str:
         if run_type != DagRunType.ASSET_TRIGGERED:
             return self.timetable.generate_run_id(run_type=run_type, **kwargs)
         return super().generate_run_id(run_type=run_type, **kwargs)
+
+
+class AssetAndTimeSchedule(Timetable):
+    """
+    Time-based schedule that waits for required assets before starting.
+
+    This timetable composes a time-based timetable with an asset condition. It
+    schedules runs according to the provided ``timetable`` (e.g. cron), but a
+    queued run will only start when all required assets are present. Unlike
+    :class:`AssetOrTimeSchedule`, this does not create asset-triggered runs.
+    """
+
+    def __init__(
+        self,
+        *,
+        timetable: Timetable,
+        assets: Collection[SerializedAsset] | SerializedAssetBase,
+    ) -> None:
+        self.timetable = timetable
+
+        if isinstance(assets, SerializedAssetBase | BaseAsset):
+            self.asset_condition = ensure_serialized_asset(assets)
+        else:
+            self.asset_condition = SerializedAssetAll([ensure_serialized_asset(a) for a in assets])
+
+        self.description = f"Triggered by assets and {timetable.description}"
+        self.periodic = timetable.periodic
+        self.can_be_scheduled = timetable.can_be_scheduled
+        self.active_runs_limit = timetable.active_runs_limit
+
+    @classmethod
+    def deserialize(cls, data: dict[str, typing.Any]) -> Timetable:
+        from airflow.serialization.decoders import decode_asset_like, decode_timetable
+
+        return cls(
+            assets=decode_asset_like(data["asset_condition"]),
+            timetable=decode_timetable(data["timetable"]),
+        )
+
+    def serialize(self) -> dict[str, typing.Any]:
+        from airflow.serialization.encoders import encode_asset_like, encode_timetable
+
+        return {
+            "asset_condition": encode_asset_like(self.asset_condition),
+            "timetable": encode_timetable(self.timetable),
+        }
+
+    def validate(self) -> None:
+        if isinstance(self.timetable, AssetTriggeredTimetable):
+            raise AirflowTimetableInvalid("cannot nest asset timetables")
+        if not isinstance(self.asset_condition, SerializedAssetBase):
+            raise AirflowTimetableInvalid("all elements in 'assets' must be assets")
+
+    @property
+    def summary(self) -> str:
+        return f"Asset and {self.timetable.summary}"
+
+    def infer_manual_data_interval(self, *, run_after: pendulum.DateTime) -> DataInterval:
+        return self.timetable.infer_manual_data_interval(run_after=run_after)
+
+    def next_dagrun_info(
+        self, *, last_automated_data_interval: DataInterval | None, restriction: TimeRestriction
+    ) -> DagRunInfo | None:
+        return self.timetable.next_dagrun_info(
+            last_automated_data_interval=last_automated_data_interval,
+            restriction=restriction,
+        )
+
+    def generate_run_id(self, *, run_type: DagRunType, **kwargs: typing.Any) -> str:
+        # All run IDs are delegated to the wrapped timetable; this class
+        # intentionally does not create ASSET_TRIGGERED runs.
+        return self.timetable.generate_run_id(run_type=run_type, **kwargs)