diff --git a/airflow-core/docs/administration-and-deployment/listeners.rst b/airflow-core/docs/administration-and-deployment/listeners.rst index 70dde0b7fd2af..70f972301731d 100644 --- a/airflow-core/docs/administration-and-deployment/listeners.rst +++ b/airflow-core/docs/administration-and-deployment/listeners.rst @@ -64,6 +64,23 @@ Beginning with Airflow 3, listeners are also notified whenever a state change is :start-after: [START howto_listen_dagrun_failure_task] :end-before: [END howto_listen_dagrun_failure_task] +Dag Skipped Interval Events +--------------------------- + +When a Dag with ``catchup=False`` skips one or more scheduled data intervals (for example after +a scheduler restart or when a paused Dag is re-enabled), the scheduler invokes the +``on_intervals_skipped`` listener hook with a +:class:`~airflow.timetables.base.SkippedIntervalsSummary`. This is the listener counterpart to the Dag-level +``on_skipped_intervals_callback``; listeners run synchronously in the scheduler, while the callback +is dispatched to the dag processor. + +- ``on_intervals_skipped`` + +.. exampleinclude:: /../src/airflow/example_dags/plugins/event_listener.py + :language: python + :start-after: [START howto_listen_intervals_skipped] + :end-before: [END howto_listen_intervals_skipped] + TaskInstance State Change Events -------------------------------- @@ -207,3 +224,5 @@ List of changes in the listener interfaces since 2.8.0 when they were introduced +-----------------+--------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------+ | 3.2.0 | ``on_task_instance_skipped`` | New listener method added to the interface | +-----------------+--------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------+ +| 3.3.0 | ``on_intervals_skipped`` | New listener method added; fires when a Dag with ``catchup=False`` skips scheduled intervals | ++-----------------+--------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------+ diff --git a/airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst b/airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst index 9b15c8f6c09f5..2ace0a8f31706 100644 --- a/airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst +++ b/airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst @@ -67,6 +67,9 @@ Name Description is not started to be executed because of a preceding branching decision in the Dag or a trigger rule which causes execution to skip so that the task execution is never scheduled. +``on_skipped_intervals_callback`` Invoked when a Dag with ``catchup=False`` skips one or more Dag + scheduled data intervals without creating Dag runs (for example + after a scheduler restart or when a paused Dag is re-enabled). =========================================== ======================================================================= ================= @@ -95,6 +98,55 @@ For example, a timeout may be caused by a number of stalling tasks, but only one Before Airflow 3.2.0, the rules above did not apply and the task instance passed to Dag callback was not related to Dag state, rather being selected as the latest task in the Dag lexicographically. +Skipped Intervals Callback +^^^^^^^^^^^^^^^^^^^^^^^^^^ + +When a Dag is defined with ``catchup=False``, the scheduler may advance past missed scheduled +intervals without creating Dag runs for them. Set ``on_skipped_intervals_callback`` on the Dag +to be notified when that happens. + +The callback is invoked once per scheduler decision that creates a new scheduled Dag run while +intervals were skipped between the previous automated run and the new run. No Dag runs are +created for the skipped intervals. + +The callback runs in the dag processor (like other Dag-level callbacks), not in the scheduler. +The context mapping contains: + +* ``dag`` -- the Dag object +* ``reason`` -- always ``"skipped_intervals"`` +* ``skipped_range`` -- a :class:`~airflow.timetables.base.DataInterval` from the + previous automated run's ``data_interval_end`` to the new run's ``data_interval_start`` + +To enumerate every skipped interval, call +:meth:`~airflow.serialization.definitions.dag.SerializedDAG.iter_dagrun_infos_between` +on ``context["dag"]`` with ``skipped_range.start`` and ``skipped_range.end``. + +Example: + +.. code-block:: python + + from airflow.sdk import DAG + from airflow.providers.standard.operators.empty import EmptyOperator + + + def log_skipped_intervals(context): + skipped_range = context["skipped_range"] + print(f"Gap: {skipped_range.start} -> {skipped_range.end}") + for info in context["dag"].iter_dagrun_infos_between(skipped_range.start, skipped_range.end): + if info.data_interval is None: + continue + interval = info.data_interval + print(f"Skipped interval: {interval.start} -> {interval.end}") + + + with DAG( + dag_id="example_skipped_intervals_callback", + schedule="@daily", + catchup=False, + on_skipped_intervals_callback=log_skipped_intervals, + ): + EmptyOperator(task_id="task") + Examples -------- diff --git a/airflow-core/docs/core-concepts/dag-run.rst b/airflow-core/docs/core-concepts/dag-run.rst index b536681036b54..940be04d861a6 100644 --- a/airflow-core/docs/core-concepts/dag-run.rst +++ b/airflow-core/docs/core-concepts/dag-run.rst @@ -112,6 +112,9 @@ If you set ``catchup=True`` in the Dag, the scheduler will kick off a Dag Run fo If your Dag is not written to handle its catchup (i.e., not limited to the interval, but instead to ``Now`` for instance.), then you will want to turn catchup off, which is the default setting or can be done explicitly by setting ``catchup=False`` in the Dag definition, if the default config has been changed for your Airflow environment. +When intervals are skipped this way, Airflow does not create Dag runs for the missed periods. +To observe those skips, set ``on_skipped_intervals_callback`` on the Dag (see :doc:`Callbacks `) or implement the ``on_intervals_skipped`` listener hook (see :doc:`Listeners `). + .. code-block:: python """ diff --git a/airflow-core/docs/howto/listener-plugin.rst b/airflow-core/docs/howto/listener-plugin.rst index c5a0acd4a6676..d1f6864603583 100644 --- a/airflow-core/docs/howto/listener-plugin.rst +++ b/airflow-core/docs/howto/listener-plugin.rst @@ -44,6 +44,7 @@ Using this plugin, following events can be listened: * Dag run is in running state. * Dag run is in success state. * Dag run is in failure state. + * scheduler skips scheduled intervals for a Dag with ``catchup=False`` * on start before event like Airflow job, scheduler * before stop for event like Airflow job, scheduler @@ -91,5 +92,12 @@ This example listens when the Dag run is change to failed state Similarly, code to listen after dag_run success and during running state can be implemented. +This example listens when the scheduler skips scheduled intervals for a Dag with ``catchup=False`` + +.. exampleinclude:: /../src/airflow/example_dags/plugins/event_listener.py + :language: python + :start-after: [START howto_listen_intervals_skipped] + :end-before: [END howto_listen_intervals_skipped] + The listener plugin files required to add the listener implementation is added as part of the Airflow plugin into ``$AIRFLOW_HOME/plugins/`` folder and loaded during Airflow startup. diff --git a/airflow-core/newsfragments/68359.feature.rst b/airflow-core/newsfragments/68359.feature.rst new file mode 100644 index 0000000000000..2aa9509400f36 --- /dev/null +++ b/airflow-core/newsfragments/68359.feature.rst @@ -0,0 +1 @@ +Add ``on_skipped_intervals_callback`` for Dags with ``catchup=False`` and the ``on_intervals_skipped`` listener hook, so operators are notified when the scheduler advances past missed scheduled intervals without creating Dag runs. diff --git a/airflow-core/src/airflow/callbacks/callback_requests.py b/airflow-core/src/airflow/callbacks/callback_requests.py index 6609dc30c6d9d..67a4a98faa875 100644 --- a/airflow-core/src/airflow/callbacks/callback_requests.py +++ b/airflow-core/src/airflow/callbacks/callback_requests.py @@ -17,6 +17,7 @@ from __future__ import annotations from collections.abc import Mapping +from datetime import datetime from typing import TYPE_CHECKING, Annotated, Any, Literal, cast import structlog @@ -26,7 +27,9 @@ from sqlalchemy.orm.attributes import set_committed_value from sqlalchemy.orm.exc import DetachedInstanceError +from airflow._shared.timezones import timezone from airflow.api_fastapi.execution_api.datamodels import taskinstance as ti_datamodel # noqa: TC001 +from airflow.timetables.base import DataInterval, SkippedIntervalsSummary from airflow.utils.state import TaskInstanceState if TYPE_CHECKING: @@ -174,8 +177,43 @@ class DagCallbackRequest(BaseCallbackRequest): type: Literal["DagCallbackRequest"] = "DagCallbackRequest" +class DagSkippedIntervalsCallbackRequest(BaseCallbackRequest): + """Store skipped intervals callback data for execution by the Dag processor.""" + + dag_id: str + skipped_range: tuple[datetime, datetime] + """Envelope from the previous automated run's data_interval_end to the new run's start.""" + type: Literal["DagSkippedIntervalsCallbackRequest"] = "DagSkippedIntervalsCallbackRequest" + + @classmethod + def from_summary( + cls, + *, + filepath: str, + bundle_name: str, + bundle_version: str | None, + dag_id: str, + summary: SkippedIntervalsSummary, + ) -> Self: + return cls( + filepath=filepath, + bundle_name=bundle_name, + bundle_version=bundle_version, + dag_id=dag_id, + skipped_range=(summary.skipped_range.start, summary.skipped_range.end), + ) + + def to_summary(self) -> SkippedIntervalsSummary: + return SkippedIntervalsSummary( + skipped_range=DataInterval( + start=timezone.coerce_datetime(self.skipped_range[0]), + end=timezone.coerce_datetime(self.skipped_range[1]), + ), + ) + + CallbackRequest = Annotated[ - DagCallbackRequest | TaskCallbackRequest | EmailRequest, + DagCallbackRequest | DagSkippedIntervalsCallbackRequest | TaskCallbackRequest | EmailRequest, Field(discriminator="type"), ] diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index ca7539131f60e..a58466eaeae62 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -32,6 +32,7 @@ from airflow.callbacks.callback_requests import ( CallbackRequest, DagCallbackRequest, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) @@ -104,7 +105,7 @@ from airflow.api_fastapi.execution_api.app import InProcessExecutionAPI from airflow.sdk.api.client import Client from airflow.sdk.bases.operator import BaseOperator - from airflow.sdk.definitions.context import Context + from airflow.sdk.definitions.context import Context, SkippedIntervalsCallbackContext from airflow.sdk.definitions.dag import DAG from airflow.sdk.definitions.mappedoperator import MappedOperator from airflow.typing_compat import Self @@ -344,6 +345,8 @@ def _execute_callbacks( _execute_task_callbacks(dagbag, request, log) elif isinstance(request, DagCallbackRequest): _execute_dag_callbacks(dagbag, request, log) + elif isinstance(request, DagSkippedIntervalsCallbackRequest): + _execute_dag_skipped_intervals_callback(dagbag, request, log) elif isinstance(request, EmailRequest): _execute_email_callbacks(dagbag, request, log) @@ -405,6 +408,32 @@ def _execute_dag_callbacks(dagbag: DagBag, request: DagCallbackRequest, log: Fil ) +def _execute_dag_skipped_intervals_callback( + dagbag: DagBag, request: DagSkippedIntervalsCallbackRequest, log: FilteringBoundLogger +) -> None: + dag, _ = _get_dag_with_task(dagbag, request.dag_id) + callbacks = dag.on_skipped_intervals_callback + if not callbacks: + log.warning("Skipped intervals callback requested, but dag didn't have any", dag_id=request.dag_id) + return + + callbacks = callbacks if isinstance(callbacks, list) else [callbacks] + summary = request.to_summary() + context: SkippedIntervalsCallbackContext = { + "dag": dag, + "reason": "skipped_intervals", + "skipped_range": summary.skipped_range, + } + + for callback in callbacks: + log.info("Executing on_skipped_intervals_callback", dag_id=request.dag_id) + try: + callback(context) + except Exception: + log.exception("Callback failed", dag_id=request.dag_id) + stats.incr("dag.callback_exceptions", tags={"dag_id": request.dag_id}) + + def _execute_task_callbacks(dagbag: DagBag, request: TaskCallbackRequest, log: FilteringBoundLogger) -> None: if not request.is_failure_callback: log.warning( diff --git a/airflow-core/src/airflow/example_dags/plugins/event_listener.py b/airflow-core/src/airflow/example_dags/plugins/event_listener.py index 91af9f5ccc6df..2a90d57534125 100644 --- a/airflow-core/src/airflow/example_dags/plugins/event_listener.py +++ b/airflow-core/src/airflow/example_dags/plugins/event_listener.py @@ -222,3 +222,23 @@ def on_dag_run_running(dag_run: DagRun, msg: str): # [END howto_listen_dagrun_running_task] + + +# [START howto_listen_intervals_skipped] +@hookimpl +def on_intervals_skipped(dag_id: str, summary): + """ + Called when a Dag with catchup=False skips one or more scheduled intervals. + + ``summary`` is a :class:`~airflow.timetables.base.SkippedIntervalsSummary` with + ``skipped_range``. Call + :meth:`~airflow.serialization.definitions.dag.SerializedDAG.iter_dagrun_infos_between` + with ``skipped_range.start`` and ``skipped_range.end`` if the full list of skipped + intervals is required. + """ + print( + f"Dag {dag_id} skipped intervals in range: {summary.skipped_range.start} -> {summary.skipped_range.end}" + ) + + +# [END howto_listen_intervals_skipped] diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index daef05b2b7c07..c48dac91e71a9 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -31,7 +31,7 @@ from datetime import date, datetime, timedelta from functools import lru_cache, partial from itertools import groupby -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, NamedTuple, cast from uuid import UUID from sqlalchemy import ( @@ -61,6 +61,7 @@ from airflow.assets.evaluation import AssetEvaluator from airflow.callbacks.callback_requests import ( DagCallbackRequest, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) @@ -71,6 +72,7 @@ from airflow.executors.executor_loader import ExecutorLoader from airflow.jobs.base_job_runner import BaseJobRunner from airflow.jobs.job import Job, JobState, perform_heartbeat +from airflow.listeners.listener import get_listener_manager from airflow.models import Deadline, Log from airflow.models.asset import ( AssetActive, @@ -113,7 +115,12 @@ 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.base import Timetable, compute_rollup_fingerprint +from airflow.timetables.base import ( + DagRunInfo, + SkippedIntervalsSummary, + Timetable, + compute_rollup_fingerprint, +) from airflow.timetables.simple import AssetTriggeredTimetable from airflow.triggers.base import TriggerEvent from airflow.utils.event_scheduler import EventScheduler @@ -151,6 +158,14 @@ DR = DagRun DM = DagModel + +class CreateDagRunsNotifications(NamedTuple): + """Skipped-interval notifications collected during Dag run creation.""" + + skip_callback_requests: list[DagSkippedIntervalsCallbackRequest] + skipped_intervals_listener_events: list[tuple[str, SkippedIntervalsSummary]] + + TASK_STUCK_IN_QUEUED_RESCHEDULE_EVENT = "stuck in queued reschedule" """:meta private:""" @@ -1978,9 +1993,10 @@ def _do_scheduling(self, session: Session) -> int: :return: Number of TIs enqueued in this iteration """ # Put a check in place to make sure we don't commit unexpectedly + dag_runs_notifications = CreateDagRunsNotifications([], []) with prohibit_commit(session) as guard: if self._scheduler_use_job_schedule: - self._create_dagruns_for_dags(guard, session) + dag_runs_notifications = self._create_dagruns_for_dags(guard, session) self._start_queued_dagruns(session) guard.commit() @@ -2017,6 +2033,11 @@ def _do_scheduling(self, session: Session) -> int: else: self.log.error("DAG '%s' not found in serialized_dag table", dag_run.dag_id) + # Dispatch skipped-intervals listener hooks and Dag callbacks after commit. + self._notify_skipped_intervals_listeners(dag_runs_notifications.skipped_intervals_listener_events) + for skip_req in dag_runs_notifications.skip_callback_requests: + self.executor.send_callback(skip_req) + with prohibit_commit(session) as guard: # Without this, the session has an invalid view of the DB session.expunge_all() @@ -2426,7 +2447,9 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st return partition_dag_ids @retry_db_transaction - def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Session) -> None: + def _create_dagruns_for_dags( + self, guard: CommitProhibitorGuard, session: Session + ) -> CreateDagRunsNotifications: """Find Dag Models needing DagRuns and Create Dag Runs with retries in case of OperationalError.""" partition_dag_ids: set[str] = self._create_dagruns_for_partitioned_asset_dags(session) @@ -2440,7 +2463,7 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio # filter asset partition triggered Dags if d.dag_id not in partition_dag_ids } - self._create_dag_runs(non_asset_dags, session) + dag_runs_notifications = self._create_dag_runs(non_asset_dags, session) if asset_triggered_dags: self._create_dag_runs_asset_triggered( dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids], @@ -2451,6 +2474,8 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio guard.commit() # END: create dagruns + return dag_runs_notifications + @provide_session def _mark_backfills_complete(self, *, session: Session = NEW_SESSION) -> None: """Mark completed backfills as completed.""" @@ -2482,8 +2507,58 @@ def _mark_backfills_complete(self, *, session: Session = NEW_SESSION) -> None: for b in backfills: b.completed_at = now - def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -> None: - """Create a DAG run and update the dag_model to control if/when the next DAGRun should be created.""" + def _collect_skipped_intervals( + self, + serdag: SerializedDAG, + new_info: DagRunInfo, + session: Session, + *, + listener_has_impls: bool, + ) -> SkippedIntervalsSummary | None: + """ + Summarize intervals skipped due to catchup=False. + + Asks the timetable whether the previous automated DagRun's immediate + successor (with catchup enabled) is earlier than the new run. Returns + ``None`` when there is no schedulable gap or when no previous run exists. + """ + if serdag.catchup: + return None + if new_info.data_interval is None: + return None + if not serdag.has_on_skipped_intervals_callback and not listener_has_impls: + return None + + prev_run = session.scalar( + select(DagRun) + .where( + DagRun.dag_id == serdag.dag_id, + DagRun.run_type == DagRunType.SCHEDULED, + DagRun.data_interval_end.is_not(None), + DagRun.data_interval_end <= new_info.data_interval.start, + ) + .order_by(DagRun.data_interval_end.desc()) + .limit(1) + ) + if prev_run is None or prev_run.data_interval_end is None: + return None + + prev_info = serdag.timetable.run_info_from_dag_run(dag_run=prev_run) + return serdag.summarize_skipped_intervals_between(prev_info, new_info) + + def _create_dag_runs( + self, dag_models: Collection[DagModel], session: Session + ) -> CreateDagRunsNotifications: + """ + Create a Dag run and update the dag_model to control if/when the next DagRun should be created. + + Returns skipped-interval notifications to dispatch after commit. + """ + skip_callback_requests: list[DagSkippedIntervalsCallbackRequest] = [] + skipped_intervals_listener_events: list[tuple[str, SkippedIntervalsSummary]] = [] + listener_has_impls = bool( + get_listener_manager().hook.on_intervals_skipped.get_hookimpls() # type: ignore[attr-defined] + ) # Bulk Fetch DagRuns with dag_id and logical_date same # as DagModel.dag_id and DagModel.next_dagrun # This list is used to verify if the DagRun already exist so that we don't attempt to create @@ -2604,6 +2679,27 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) - active_non_backfill_runs=active_runs_of_dags[dag_model.dag_id], ) + if data_interval is not None: + skipped_summary = self._collect_skipped_intervals( + serdag=serdag, + new_info=next_info, + session=session, + listener_has_impls=listener_has_impls, + ) + if skipped_summary is not None: + if listener_has_impls: + skipped_intervals_listener_events.append((serdag.dag_id, skipped_summary)) + if serdag.has_on_skipped_intervals_callback: + skip_callback_requests.append( + DagSkippedIntervalsCallbackRequest.from_summary( + filepath=dag_model.relative_fileloc or "", + bundle_name=dag_model.bundle_name, + bundle_version=created_run.bundle_version, + dag_id=serdag.dag_id, + summary=skipped_summary, + ) + ) + # Exceptions like ValueError, ParamValidationError, etc. are raised by # DagModel.create_dagrun() when dag is misconfigured. The scheduler should not # crash due to misconfigured dags. We should log any exception encountered @@ -2618,6 +2714,27 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) - # TODO[HA]: Should we do a session.flush() so we don't have to keep lots of state/object in # memory for larger dags? or expunge_all() + return CreateDagRunsNotifications( + skip_callback_requests=skip_callback_requests, + skipped_intervals_listener_events=skipped_intervals_listener_events, + ) + + def _notify_skipped_intervals_listeners( + self, + events: list[tuple[str, SkippedIntervalsSummary]], + ) -> None: + for dag_id, summary in events: + try: + get_listener_manager().hook.on_intervals_skipped( # type: ignore[attr-defined] + dag_id=dag_id, + summary=summary, + ) + except Exception: + self.log.exception( + "Error notifying listener of skipped intervals", + dag_id=dag_id, + ) + def _create_dag_runs_asset_triggered( self, *, diff --git a/airflow-core/src/airflow/listeners/listener.py b/airflow-core/src/airflow/listeners/listener.py index 06be7a1b9b908..798312a0a6c4a 100644 --- a/airflow-core/src/airflow/listeners/listener.py +++ b/airflow-core/src/airflow/listeners/listener.py @@ -21,7 +21,7 @@ from airflow._shared.listeners.listener import ListenerManager from airflow._shared.listeners.spec import lifecycle, taskinstance -from airflow.listeners.spec import asset, dagrun, importerrors +from airflow.listeners.spec import asset, dag, dagrun, importerrors from airflow.plugins_manager import integrate_listener_plugins @@ -32,6 +32,7 @@ def get_listener_manager() -> ListenerManager: Registers the following listeners: - lifecycle: on_starting, before_stopping + - dag: on_intervals_skipped - dagrun: on_dag_run_running, on_dag_run_success, on_dag_run_failed - taskinstance: on_task_instance_running, on_task_instance_success, etc. - asset: on_asset_created, on_asset_changed, etc. @@ -40,6 +41,7 @@ def get_listener_manager() -> ListenerManager: _listener_manager = ListenerManager() _listener_manager.add_hookspecs(lifecycle) + _listener_manager.add_hookspecs(dag) _listener_manager.add_hookspecs(dagrun) _listener_manager.add_hookspecs(taskinstance) _listener_manager.add_hookspecs(asset) diff --git a/airflow-core/src/airflow/listeners/spec/dag.py b/airflow-core/src/airflow/listeners/spec/dag.py new file mode 100644 index 0000000000000..51f5ad7c09845 --- /dev/null +++ b/airflow-core/src/airflow/listeners/spec/dag.py @@ -0,0 +1,31 @@ +# 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 + +from typing import TYPE_CHECKING + +from pluggy import HookspecMarker + +if TYPE_CHECKING: + from airflow.timetables.base import SkippedIntervalsSummary + +hookspec = HookspecMarker("airflow") + + +@hookspec +def on_intervals_skipped(dag_id: str, summary: SkippedIntervalsSummary): + """Execute when a Dag with catchup=False skips one or more scheduled intervals.""" diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py b/airflow-core/src/airflow/serialization/definitions/dag.py index 8ed0fee2ccabd..89477450c45c6 100644 --- a/airflow-core/src/airflow/serialization/definitions/dag.py +++ b/airflow-core/src/airflow/serialization/definitions/dag.py @@ -55,7 +55,7 @@ from airflow.serialization.definitions.deadline import DeadlineAlertFields, SerializedReferenceModels from airflow.serialization.definitions.param import SerializedParamsDict from airflow.serialization.enums import DagAttributeTypes as DAT, Encoding -from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction +from airflow.timetables.base import DagRunInfo, DataInterval, SkippedIntervalsSummary, TimeRestriction from airflow.utils.helpers import prune_dict from airflow.utils.session import NEW_SESSION, provide_session from airflow.utils.state import DagRunState, TaskInstanceState @@ -122,6 +122,7 @@ class SerializedDAG: fail_fast: bool = False has_on_failure_callback: bool = False has_on_success_callback: bool = False + has_on_skipped_intervals_callback: bool = False is_paused_upon_creation: bool | None = None max_active_runs: int = 16 max_active_tasks: int = 16 @@ -523,6 +524,48 @@ def iter_dagrun_infos_between( dag_id=self.dag_id, ) + def summarize_skipped_intervals_between( + self, + prev_info: DagRunInfo, + new_info: DagRunInfo, + ) -> SkippedIntervalsSummary | None: + """ + Summarize intervals skipped between two automated Dag runs. + + Asks the timetable for the immediate successor of *prev_info* with + catchup enabled. Returns a summary only when that expected successor's + logical date is strictly earlier than *new_info*'s logical date — i.e. + when at least one scheduled run was skipped. + + Returns ``None`` when there is no schedulable gap (including consecutive + zero-width trigger-timetable runs, where wall-clock endpoints differ + but the expected next logical date matches the new run). + """ + if prev_info.data_interval is None or new_info.data_interval is None: + return None + if new_info.logical_date is None: + return None + + prev_interval_end = prev_info.data_interval.end + new_interval_start = new_info.data_interval.start + if prev_interval_end >= new_interval_start: + return None + + expected = self.next_dagrun_info(last_automated_run_info=prev_info, restricted=False) + if expected is None or expected.logical_date is None: + return None + if expected.logical_date >= new_info.logical_date: + return None + + from airflow._shared.timezones import timezone + + return SkippedIntervalsSummary( + skipped_range=DataInterval( + start=timezone.coerce_datetime(prev_interval_end), + end=timezone.coerce_datetime(new_interval_start), + ), + ) + @provide_session def get_concurrency_reached(self, *, session=NEW_SESSION) -> bool: """Return a boolean indicating whether the max_active_tasks limit for this DAG has been reached.""" diff --git a/airflow-core/src/airflow/serialization/schema.json b/airflow-core/src/airflow/serialization/schema.json index 872c3a1331ee3..c9a3d22159570 100644 --- a/airflow-core/src/airflow/serialization/schema.json +++ b/airflow-core/src/airflow/serialization/schema.json @@ -217,6 +217,7 @@ "is_paused_upon_creation": { "type": "boolean" }, "has_on_success_callback": { "type": "boolean", "default": false }, "has_on_failure_callback": { "type": "boolean", "default": false }, + "has_on_skipped_intervals_callback": { "type": "boolean", "default": false }, "render_template_as_native_obj": { "type": "boolean", "default": false }, "tags": { "type": "array" }, "task_group": {"anyOf": [ diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py index 54bc3389c64ce..513d1de5d0723 100644 --- a/airflow-core/src/airflow/serialization/serialized_objects.py +++ b/airflow-core/src/airflow/serialization/serialized_objects.py @@ -1729,6 +1729,8 @@ def serialize_dag(cls, dag: DAG) -> dict: serialized_dag["has_on_success_callback"] = True if dag.has_on_failure_callback: serialized_dag["has_on_failure_callback"] = True + if dag.has_on_skipped_intervals_callback: + serialized_dag["has_on_skipped_intervals_callback"] = True # TODO: Move this logic to a better place -- ideally before serializing contents of default_args. # There is some duplication with this and SerializedBaseOperator.partial_kwargs serialization. @@ -1855,6 +1857,8 @@ def _deserialize_dag_internal( dag.has_on_success_callback = True if "has_on_failure_callback" in encoded_dag: dag.has_on_failure_callback = True + if "has_on_skipped_intervals_callback" in encoded_dag: + dag.has_on_skipped_intervals_callback = True dag.deadline = encoded_dag.get("deadline") @@ -2211,6 +2215,7 @@ class LazyDeserializedDAG(pydantic.BaseModel): "dag_display_name", "has_on_success_callback", "has_on_failure_callback", + "has_on_skipped_intervals_callback", "tags", # Attr properties that are nullable, or have a default that loads from config "description", diff --git a/airflow-core/src/airflow/timetables/base.py b/airflow-core/src/airflow/timetables/base.py index 365f980b93274..cb76281e08a5c 100644 --- a/airflow-core/src/airflow/timetables/base.py +++ b/airflow-core/src/airflow/timetables/base.py @@ -75,6 +75,20 @@ def exact(cls, at: DateTime) -> DataInterval: return cls(start=at, end=at) +class SkippedIntervalsSummary(NamedTuple): + """ + Summary of scheduled data intervals skipped due to ``catchup=False``. + + ``skipped_range`` spans from the previous automated Dag run's + ``data_interval_end`` to the new run's ``data_interval_start``. + Call :meth:`~airflow.serialization.definitions.dag.SerializedDAG.iter_dagrun_infos_between` + with ``skipped_range.start`` and ``skipped_range.end`` in a callback if the full sequence + is required. + """ + + skipped_range: DataInterval + + class TimeRestriction(NamedTuple): """ Restriction on when a DAG can be scheduled for a run. diff --git a/airflow-core/tests/unit/callbacks/test_callback_requests.py b/airflow-core/tests/unit/callbacks/test_callback_requests.py index 9b85d61e5dba0..6cb7f24c70373 100644 --- a/airflow-core/tests/unit/callbacks/test_callback_requests.py +++ b/airflow-core/tests/unit/callbacks/test_callback_requests.py @@ -31,12 +31,14 @@ CallbackRequest, DagCallbackRequest, DagRunContext, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) from airflow.models import DagRun from airflow.models.taskinstance import TaskInstance from airflow.serialization.definitions.baseoperator import SerializedBaseOperator +from airflow.timetables.base import DataInterval, SkippedIntervalsSummary from airflow.utils.state import State, TaskInstanceState pytestmark = pytest.mark.db_test @@ -400,6 +402,46 @@ def test_dag_callback_request_serialization_with_context(self): assert result.context_from_server.last_ti.task_id == "test_task" +class TestDagSkippedIntervalsCallbackRequest: + def test_to_json_roundtrip(self): + interval_start = timezone.datetime(2024, 1, 1) + interval_end = timezone.datetime(2024, 1, 2) + summary = SkippedIntervalsSummary( + skipped_range=DataInterval(start=interval_start, end=interval_end), + ) + request = DagSkippedIntervalsCallbackRequest.from_summary( + filepath="test.py", + dag_id="test_dag", + bundle_name="testing", + bundle_version="v1", + summary=summary, + ) + + result = DagSkippedIntervalsCallbackRequest.from_json(request.to_json()) + + assert result == request + assert result.to_summary() == summary + + def test_callback_request_union_with_skipped_intervals(self): + interval_start = timezone.datetime(2024, 1, 1) + interval_end = timezone.datetime(2024, 1, 2) + request_data = { + "type": "DagSkippedIntervalsCallbackRequest", + "filepath": "test.py", + "dag_id": "test_dag", + "bundle_name": "testing", + "bundle_version": "v1", + "skipped_range": [interval_start.isoformat(), interval_end.isoformat()], + } + + adapter = TypeAdapter(CallbackRequest) + callback_request = adapter.validate_python(request_data) + + assert isinstance(callback_request, DagSkippedIntervalsCallbackRequest) + assert callback_request.dag_id == "test_dag" + assert callback_request.skipped_range == (interval_start, interval_end) + + class TestEmailRequest: def test_email_notification_request_serialization(self): """Test EmailRequest can be serialized and used in CallbackRequest union.""" diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py b/airflow-core/tests/unit/dag_processing/test_processor.py index dd7f603b35bb0..aa256302e90d6 100644 --- a/airflow-core/tests/unit/dag_processing/test_processor.py +++ b/airflow-core/tests/unit/dag_processing/test_processor.py @@ -46,6 +46,7 @@ CallbackRequest, DagCallbackRequest, DagRunContext, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) @@ -59,6 +60,7 @@ ToManager, _execute_callbacks, _execute_dag_callbacks, + _execute_dag_skipped_intervals_callback, _execute_email_callbacks, _execute_task_callbacks, _parse_file, @@ -85,6 +87,7 @@ XComSequenceSliceResult, ) from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance +from airflow.timetables.base import DataInterval, SkippedIntervalsSummary from airflow.utils.session import create_session from airflow.utils.state import TaskInstanceState @@ -93,6 +96,30 @@ if TYPE_CHECKING: from kgb import SpyAgency + +def _skipped_intervals_callback_request( + *, + filepath: str = "test.py", + dag_id: str = "test_dag", + bundle_name: str = "testing", + bundle_version: str | None = None, + interval_start=None, + interval_end=None, +) -> DagSkippedIntervalsCallbackRequest: + interval_start = interval_start or timezone.utcnow() + interval_end = interval_end or timezone.utcnow() + summary = SkippedIntervalsSummary( + skipped_range=DataInterval(start=interval_start, end=interval_end), + ) + return DagSkippedIntervalsCallbackRequest.from_summary( + filepath=filepath, + dag_id=dag_id, + bundle_name=bundle_name, + bundle_version=bundle_version, + summary=summary, + ) + + pytestmark = pytest.mark.db_test @@ -814,6 +841,22 @@ def test_execute_callbacks_locks_bundle_version(self): mock_lock.return_value.__exit__.assert_called_once() mock_execute.assert_called_once_with(dagbag, callbacks[0], log) + def test_execute_callbacks_routes_skipped_intervals_request(self): + callbacks = [ + _skipped_intervals_callback_request(bundle_version="some_commit_hash"), + ] + log = MagicMock(spec=FilteringBoundLogger) + dagbag = MagicMock(spec=DagBag) + + with ( + patch("airflow.dag_processing.processor.BundleVersionLock") as mock_lock, + patch("airflow.dag_processing.processor._execute_dag_skipped_intervals_callback") as mock_execute, + ): + _execute_callbacks(dagbag, callbacks, log) + + mock_lock.assert_called_once_with(bundle_name="testing", bundle_version="some_commit_hash") + mock_execute.assert_called_once_with(dagbag, callbacks[0], log) + class TestExecuteDagCallbacks: """Test the _execute_dag_callbacks function with context_from_server""" @@ -1308,6 +1351,100 @@ def fake_collect_dags(self, *args, **kwargs): assert operation_response in caplog.text +class TestExecuteDagSkippedIntervalsCallback: + def test_execute_skipped_intervals_callback(self, spy_agency): + context_received = None + + def on_skipped_intervals(context): + nonlocal context_received + context_received = context + + with DAG(dag_id="test_dag", on_skipped_intervals_callback=on_skipped_intervals) as dag: + BaseOperator(task_id="test_task") + + def fake_collect_dags(self, *args, **kwargs): + self.dags[dag.dag_id] = dag + + spy_agency.spy_on(DagBag.collect_dags, call_fake=fake_collect_dags, owner=DagBag) + + dagbag = DagBag() + dagbag.collect_dags() + + interval_start = timezone.datetime(2024, 1, 1) + interval_end = timezone.datetime(2024, 1, 2) + request = _skipped_intervals_callback_request( + interval_start=interval_start, + interval_end=interval_end, + ) + + log = structlog.get_logger() + _execute_dag_skipped_intervals_callback(dagbag, request, log) + + assert context_received is not None + assert context_received["dag"] is dag + assert context_received["reason"] == "skipped_intervals" + assert context_received["skipped_range"] == DataInterval(start=interval_start, end=interval_end) + + def test_execute_skipped_intervals_callback_multiple_callbacks(self, spy_agency): + call_count = 0 + + def on_skipped_1(context): + nonlocal call_count + call_count += 1 + + def on_skipped_2(context): + nonlocal call_count + call_count += 1 + + with DAG( + dag_id="test_dag", + on_skipped_intervals_callback=[on_skipped_1, on_skipped_2], + ) as dag: + BaseOperator(task_id="test_task") + + def fake_collect_dags(self, *args, **kwargs): + self.dags[dag.dag_id] = dag + + spy_agency.spy_on(DagBag.collect_dags, call_fake=fake_collect_dags, owner=DagBag) + + dagbag = DagBag() + dagbag.collect_dags() + + request = _skipped_intervals_callback_request() + + _execute_dag_skipped_intervals_callback(dagbag, request, structlog.get_logger()) + assert call_count == 2 + + def test_execute_skipped_intervals_callback_no_callback_defined(self, spy_agency): + with DAG(dag_id="test_dag") as dag: + BaseOperator(task_id="test_task") + + def fake_collect_dags(self, *args, **kwargs): + self.dags[dag.dag_id] = dag + + spy_agency.spy_on(DagBag.collect_dags, call_fake=fake_collect_dags, owner=DagBag) + + dagbag = DagBag() + dagbag.collect_dags() + + request = _skipped_intervals_callback_request() + + log = MagicMock(spec=FilteringBoundLogger) + _execute_dag_skipped_intervals_callback(dagbag, request, log) + + log.warning.assert_called_once_with( + "Skipped intervals callback requested, but dag didn't have any", + dag_id="test_dag", + ) + + def test_execute_skipped_intervals_callback_missing_dag(self): + dagbag = DagBag() + request = _skipped_intervals_callback_request(dag_id="missing_dag") + + with pytest.raises(ValueError, match="DAG 'missing_dag' not found in DagBag"): + _execute_dag_skipped_intervals_callback(dagbag, request, structlog.get_logger()) + + class TestExecuteTaskCallbacks: """Test the _execute_task_callbacks function""" diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index 4add965a6abda..bd261a839be4a 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -49,6 +49,7 @@ from airflow.callbacks.callback_requests import ( DagCallbackRequest, DagRunContext, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) @@ -10183,6 +10184,323 @@ def test_do_scheduling_multi_team_schedules_task_instances(self, dag_maker, sess session.refresh(ti) assert ti.state != State.NONE + def test_collect_skipped_intervals_returns_empty_when_catchup_true(self, session, dag_maker): + with dag_maker( + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=True, + on_skipped_intervals_callback=lambda ctx: None, + session=session, + ): + EmptyOperator(task_id="dummy") + + serdag = dag_maker.serialized_dag + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec]) + new_info = DagRunInfo.interval( + DEFAULT_DATE + timedelta(days=4), + DEFAULT_DATE + timedelta(days=5), + ) + assert ( + self.job_runner._collect_skipped_intervals(serdag, new_info, session, listener_has_impls=False) + is None + ) + + def test_collect_skipped_intervals_returns_empty_without_callback_or_listener(self, session, dag_maker): + with dag_maker( + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + session=session, + ): + EmptyOperator(task_id="dummy") + + prev_start = DEFAULT_DATE + prev_end = DEFAULT_DATE + timedelta(days=1) + dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=prev_start, + data_interval=(prev_start, prev_end), + state=State.SUCCESS, + ) + + serdag = dag_maker.serialized_dag + assert not serdag.has_on_skipped_intervals_callback + new_info = DagRunInfo.interval( + DEFAULT_DATE + timedelta(days=4), + DEFAULT_DATE + timedelta(days=5), + ) + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec]) + assert ( + self.job_runner._collect_skipped_intervals(serdag, new_info, session, listener_has_impls=False) + is None + ) + + def test_collect_skipped_intervals_returns_gap_summary(self, session, dag_maker): + with dag_maker( + dag_id="test_collect_skipped_intervals", + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + on_skipped_intervals_callback=lambda ctx: None, + session=session, + ): + EmptyOperator(task_id="dummy") + + prev_start = DEFAULT_DATE + prev_end = DEFAULT_DATE + timedelta(days=1) + dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=prev_start, + data_interval=(prev_start, prev_end), + state=State.SUCCESS, + ) + + new_start = DEFAULT_DATE + timedelta(days=4) + new_end = DEFAULT_DATE + timedelta(days=5) + serdag = dag_maker.serialized_dag + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec]) + + summary = self.job_runner._collect_skipped_intervals( + serdag, + DagRunInfo.interval(new_start, new_end), + session, + listener_has_impls=False, + ) + + assert summary is not None + assert summary.skipped_range == DataInterval(start=prev_end, end=new_start) + + def test_collect_skipped_intervals_returns_empty_without_previous_run(self, session, dag_maker): + with dag_maker( + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + on_skipped_intervals_callback=lambda ctx: None, + session=session, + ): + EmptyOperator(task_id="dummy") + + serdag = dag_maker.serialized_dag + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec]) + new_info = DagRunInfo.interval( + DEFAULT_DATE + timedelta(days=1), + DEFAULT_DATE + timedelta(days=2), + ) + assert ( + self.job_runner._collect_skipped_intervals(serdag, new_info, session, listener_has_impls=False) + is None + ) + + def test_collect_skipped_intervals_returns_empty_when_no_gap(self, session, dag_maker): + with dag_maker( + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + on_skipped_intervals_callback=lambda ctx: None, + session=session, + ): + EmptyOperator(task_id="dummy") + + prev_start = DEFAULT_DATE + prev_end = DEFAULT_DATE + timedelta(days=1) + dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=prev_start, + data_interval=(prev_start, prev_end), + state=State.SUCCESS, + ) + + serdag = dag_maker.serialized_dag + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec]) + new_info = DagRunInfo.interval(prev_end, prev_end + timedelta(days=1)) + assert ( + self.job_runner._collect_skipped_intervals(serdag, new_info, session, listener_has_impls=False) + is None + ) + + @conf_vars({("scheduler", "create_cron_data_intervals"): "False"}) + def test_create_dag_runs_does_not_emit_skipped_intervals_for_consecutive_cron_runs( + self, session, dag_maker + ): + """Regression: consecutive CronTriggerTimetable runs must not fire skip callbacks.""" + with dag_maker( + dag_id="test_consecutive_cron_no_false_skip", + schedule="@daily", + start_date=DEFAULT_DATE, + catchup=False, + on_skipped_intervals_callback=lambda ctx: None, + session=session, + ): + EmptyOperator(task_id="dummy") + + dr1 = dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=DEFAULT_DATE, + data_interval=DataInterval.exact(DEFAULT_DATE), + state=State.SUCCESS, + ) + + serdag = dag_maker.serialized_dag + next_info = serdag.next_dagrun_info( + last_automated_run_info=serdag.timetable.run_info_from_dag_run(dag_run=dr1), + restricted=False, + ) + assert next_info is not None + dag_model = dag_maker.dag_model + dag_model.next_dagrun = next_info.logical_date + dag_model.next_dagrun_data_interval = next_info.data_interval + dag_model.next_dagrun_create_after = next_info.run_after + session.merge(dag_model) + session.commit() + + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[MockExecutor(do_update=False)]) + notifications = self.job_runner._create_dag_runs([dag_model], session) + + assert notifications.skip_callback_requests == [] + assert notifications.skipped_intervals_listener_events == [] + + def test_create_dag_runs_returns_skipped_intervals_callback_request(self, session, dag_maker): + with dag_maker( + dag_id="test_create_dag_runs_skipped_intervals_callback", + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + on_skipped_intervals_callback=lambda ctx: None, + session=session, + ) as dag: + EmptyOperator(task_id="dummy") + + prev_start = DEFAULT_DATE + prev_end = DEFAULT_DATE + timedelta(days=1) + dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=prev_start, + data_interval=(prev_start, prev_end), + state=State.SUCCESS, + ) + + new_start = DEFAULT_DATE + timedelta(days=4) + new_end = DEFAULT_DATE + timedelta(days=5) + dag_model = dag_maker.dag_model + dag_model.next_dagrun = new_start + dag_model.next_dagrun_data_interval = DataInterval(start=new_start, end=new_end) + dag_model.next_dagrun_create_after = new_end + session.merge(dag_model) + session.commit() + + scheduler_job = Job() + mock_exec = MockExecutor(do_update=False) + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[mock_exec]) + + notifications = self.job_runner._create_dag_runs([dag_model], session) + session.commit() + + skip_requests = notifications.skip_callback_requests + + assert len(skip_requests) == 1 + request = skip_requests[0] + assert isinstance(request, DagSkippedIntervalsCallbackRequest) + assert request.dag_id == dag.dag_id + assert request.filepath == dag_model.relative_fileloc + assert request.bundle_name == dag_model.bundle_name + assert request.skipped_range == (prev_end, new_start) + + def test_create_dag_runs_defers_skipped_intervals_listener_notification( + self, session, dag_maker, listener_manager + ): + from unit.listeners import skipped_intervals_listener + + listener_manager(skipped_intervals_listener) + + with dag_maker( + dag_id="test_create_dag_runs_deferred_listener", + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + session=session, + ): + EmptyOperator(task_id="dummy") + + prev_start = DEFAULT_DATE + prev_end = DEFAULT_DATE + timedelta(days=1) + dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=prev_start, + data_interval=(prev_start, prev_end), + state=State.SUCCESS, + ) + + new_start = DEFAULT_DATE + timedelta(days=4) + new_end = DEFAULT_DATE + timedelta(days=5) + dag_model = dag_maker.dag_model + dag_model.next_dagrun = new_start + dag_model.next_dagrun_data_interval = DataInterval(start=new_start, end=new_end) + dag_model.next_dagrun_create_after = new_end + session.merge(dag_model) + session.commit() + + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[MockExecutor(do_update=False)]) + + with mock.patch.object( + self.job_runner, + "_notify_skipped_intervals_listeners", + wraps=self.job_runner._notify_skipped_intervals_listeners, + ) as mock_notify: + notifications = self.job_runner._create_dag_runs([dag_model], session) + session.commit() + mock_notify.assert_not_called() + + assert len(notifications.skipped_intervals_listener_events) == 1 + self.job_runner._notify_skipped_intervals_listeners(notifications.skipped_intervals_listener_events) + assert len(skipped_intervals_listener.events) == 1 + + def test_do_scheduling_dispatches_skipped_intervals_callback(self, session, dag_maker): + with dag_maker( + dag_id="test_do_scheduling_skipped_intervals_callback", + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + on_skipped_intervals_callback=lambda ctx: None, + session=session, + ): + EmptyOperator(task_id="dummy") + + prev_start = DEFAULT_DATE + prev_end = DEFAULT_DATE + timedelta(days=1) + dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=prev_start, + data_interval=(prev_start, prev_end), + state=State.SUCCESS, + ) + + new_start = DEFAULT_DATE + timedelta(days=4) + new_end = DEFAULT_DATE + timedelta(days=5) + dag_model = dag_maker.dag_model + dag_model.next_dagrun = new_start + dag_model.next_dagrun_data_interval = DataInterval(start=new_start, end=new_end) + dag_model.next_dagrun_create_after = new_end + session.merge(dag_model) + session.commit() + + scheduler_job = Job() + mock_exec = MockExecutor(do_update=False) + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[mock_exec]) + + with mock.patch.object(mock_exec, "send_callback") as mock_send_callback: + self.job_runner._do_scheduling(session) + + mock_send_callback.assert_called_once() + request = mock_send_callback.call_args.args[0] + assert isinstance(request, DagSkippedIntervalsCallbackRequest) + @pytest.mark.need_serialized_dag def test_schedule_dag_run_with_upstream_skip(dag_maker, session): diff --git a/airflow-core/tests/unit/listeners/skipped_intervals_listener.py b/airflow-core/tests/unit/listeners/skipped_intervals_listener.py new file mode 100644 index 0000000000000..4b0f52069eb59 --- /dev/null +++ b/airflow-core/tests/unit/listeners/skipped_intervals_listener.py @@ -0,0 +1,37 @@ +# +# 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 + +from typing import TYPE_CHECKING + +from airflow.listeners import hookimpl + +if TYPE_CHECKING: + from airflow.timetables.base import SkippedIntervalsSummary + +events: list[tuple[str, SkippedIntervalsSummary]] = [] + + +@hookimpl +def on_intervals_skipped(dag_id: str, summary: SkippedIntervalsSummary): + events.append((dag_id, summary)) + + +def clear(): + global events + events = [] diff --git a/airflow-core/tests/unit/listeners/test_skipped_intervals_listener.py b/airflow-core/tests/unit/listeners/test_skipped_intervals_listener.py new file mode 100644 index 0000000000000..885070d12f27d --- /dev/null +++ b/airflow-core/tests/unit/listeners/test_skipped_intervals_listener.py @@ -0,0 +1,98 @@ +# 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 + +from datetime import timedelta +from typing import TYPE_CHECKING + +import pytest + +from airflow._shared.timezones import timezone +from airflow.jobs.job import Job +from airflow.jobs.scheduler_job_runner import SchedulerJobRunner +from airflow.models.dag import DagModel +from airflow.providers.standard.operators.empty import EmptyOperator +from airflow.timetables.base import DataInterval +from airflow.utils.state import State +from airflow.utils.types import DagRunType + +from tests_common.test_utils.mock_executor import MockExecutor +from unit.listeners import skipped_intervals_listener + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + + from tests_common.pytest_plugin import DagMaker + +pytestmark = [pytest.mark.db_test, pytest.mark.need_serialized_dag] + +DEFAULT_DATE = timezone.datetime(2016, 1, 1) + + +@pytest.fixture(autouse=True) +def clean_listener_state(): + yield + skipped_intervals_listener.clear() + + +def _configure_gap_dag_model(dag_maker: DagMaker, session: Session): + prev_start = DEFAULT_DATE + prev_end = DEFAULT_DATE + timedelta(days=1) + dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=prev_start, + data_interval=DataInterval(start=prev_start, end=prev_end), + state=State.SUCCESS, + ) + + new_start = DEFAULT_DATE + timedelta(days=4) + new_end = DEFAULT_DATE + timedelta(days=5) + dag_model = DagModel.get_dagmodel(dag_maker.dag.dag_id, session=session) + if dag_model is None: + raise RuntimeError(f"DagModel for {dag_maker.dag.dag_id} not found") + dag_model.next_dagrun = new_start + dag_model.next_dagrun_data_interval = DataInterval(start=new_start, end=new_end) + dag_model.next_dagrun_create_after = new_end + session.merge(dag_model) + session.commit() + return dag_model, prev_end, new_start + + +def test_listener_notified_when_intervals_skipped(session, dag_maker, listener_manager): + listener_manager(skipped_intervals_listener) + + with dag_maker( + dag_id="test_listener_skipped_intervals", + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + session=session, + ): + EmptyOperator(task_id="dummy") + + dag_model, prev_end, new_start = _configure_gap_dag_model(dag_maker, session) + + job_runner = SchedulerJobRunner(job=Job(), executors=[MockExecutor(do_update=False)]) + notifications = job_runner._create_dag_runs([dag_model], session) + session.commit() + job_runner._notify_skipped_intervals_listeners(notifications.skipped_intervals_listener_events) + + assert len(skipped_intervals_listener.events) == 1 + dag_id, summary = skipped_intervals_listener.events[0] + assert dag_id == dag_maker.dag.dag_id + assert summary.skipped_range.start == prev_end + assert summary.skipped_range.end == new_start diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py index 4708365e846e2..18246f3fa0a96 100644 --- a/airflow-core/tests/unit/serialization/test_dag_serialization.py +++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py @@ -1566,6 +1566,7 @@ def test_dag_serialized_fields_with_schema(self): "tasks", "has_on_success_callback", "has_on_failure_callback", + "has_on_skipped_intervals_callback", "dag_dependencies", "params", } @@ -2402,6 +2403,37 @@ def test_dag_on_failure_callback_roundtrip(self, passed_failure_callback, expect assert deserialized_dag.has_on_failure_callback is expected_value + @pytest.mark.parametrize( + ("passed_skipped_intervals_callback", "expected_value"), + [ + ({"on_skipped_intervals_callback": lambda x: print("hi")}, True), + ({}, False), + ], + ) + def test_dag_on_skipped_intervals_callback_roundtrip( + self, passed_skipped_intervals_callback, expected_value + ): + """ + Test that when on_skipped_intervals_callback is passed to the DAG, + has_on_skipped_intervals_callback is stored in Serialized JSON blob. + """ + dag = DAG( + dag_id="test_dag_on_skipped_intervals_callback_roundtrip", + schedule=None, + **passed_skipped_intervals_callback, + ) + BaseOperator(task_id="simple_task", dag=dag, start_date=datetime(2019, 8, 1)) + + serialized_dag = DagSerialization.to_dict(dag) + if expected_value: + assert "has_on_skipped_intervals_callback" in serialized_dag["dag"] + else: + assert "has_on_skipped_intervals_callback" not in serialized_dag["dag"] + + deserialized_dag = DagSerialization.from_dict(serialized_dag) + + assert deserialized_dag.has_on_skipped_intervals_callback is expected_value + @pytest.mark.parametrize( ("dag_arg", "conf_arg", "expected"), [ diff --git a/airflow-core/tests/unit/timetables/test_skipped_intervals_summary.py b/airflow-core/tests/unit/timetables/test_skipped_intervals_summary.py new file mode 100644 index 0000000000000..efe823715a928 --- /dev/null +++ b/airflow-core/tests/unit/timetables/test_skipped_intervals_summary.py @@ -0,0 +1,103 @@ +# 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 + +from datetime import timedelta + +import pendulum + +from airflow.providers.standard.operators.empty import EmptyOperator +from airflow.sdk import DAG, CronTriggerTimetable +from airflow.timetables.base import DagRunInfo, DataInterval + +from tests_common.test_utils.dag import create_scheduler_dag + +DEFAULT_DATE = pendulum.datetime(2016, 1, 1, tz="UTC") + + +def test_summarize_skipped_intervals_between_returns_gap_summary(): + dag = DAG( + dag_id="test_summarize_skipped_intervals", + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + ) + EmptyOperator(task_id="dummy", dag=dag) + scheduler_dag = create_scheduler_dag(dag) + + prev_end = DEFAULT_DATE + timedelta(days=1) + new_start = DEFAULT_DATE + timedelta(days=4) + prev_info = DagRunInfo.interval(DEFAULT_DATE, prev_end) + new_info = DagRunInfo.interval(new_start, new_start + timedelta(days=1)) + + summary = scheduler_dag.summarize_skipped_intervals_between(prev_info, new_info) + + assert summary is not None + assert summary.skipped_range == DataInterval(start=prev_end, end=new_start) + + +def test_summarize_skipped_intervals_between_returns_none_without_gap(): + dag = DAG( + dag_id="test_summarize_skipped_intervals_no_gap", + schedule=timedelta(days=1), + start_date=DEFAULT_DATE, + catchup=False, + ) + EmptyOperator(task_id="dummy", dag=dag) + scheduler_dag = create_scheduler_dag(dag) + + boundary = DEFAULT_DATE + timedelta(days=1) + prev_info = DagRunInfo.interval(DEFAULT_DATE, boundary) + new_info = DagRunInfo.interval(boundary, boundary + timedelta(days=1)) + assert scheduler_dag.summarize_skipped_intervals_between(prev_info, new_info) is None + + +def test_summarize_skipped_intervals_between_returns_none_for_consecutive_trigger_runs(): + """Zero-width consecutive CronTriggerTimetable runs must not look like a skip.""" + dag = DAG( + dag_id="test_summarize_consecutive_trigger", + schedule=CronTriggerTimetable("@daily", timezone="UTC"), + start_date=DEFAULT_DATE, + catchup=False, + ) + EmptyOperator(task_id="dummy", dag=dag) + scheduler_dag = create_scheduler_dag(dag) + + prev_info = DagRunInfo.exact(DEFAULT_DATE) + new_info = DagRunInfo.exact(DEFAULT_DATE + timedelta(days=1)) + assert scheduler_dag.summarize_skipped_intervals_between(prev_info, new_info) is None + + +def test_summarize_skipped_intervals_between_returns_gap_for_skipped_trigger_runs(): + """A real multi-period jump on CronTriggerTimetable must still report a skip.""" + dag = DAG( + dag_id="test_summarize_skipped_trigger", + schedule=CronTriggerTimetable("@daily", timezone="UTC"), + start_date=DEFAULT_DATE, + catchup=False, + ) + EmptyOperator(task_id="dummy", dag=dag) + scheduler_dag = create_scheduler_dag(dag) + + prev_info = DagRunInfo.exact(DEFAULT_DATE) + new_start = DEFAULT_DATE + timedelta(days=3) + new_info = DagRunInfo.exact(new_start) + + summary = scheduler_dag.summarize_skipped_intervals_between(prev_info, new_info) + + assert summary is not None + assert summary.skipped_range == DataInterval(start=DEFAULT_DATE, end=new_start) diff --git a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go index 77725b0ac4efc..59a59f63950c9 100644 --- a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go +++ b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go @@ -142,7 +142,7 @@ func TestPack_CrossArchExecutableWithMetadataFile(t *testing.T) { sdk: language: "go" version: "` + sdkVersion + `" - supervisor_schema_version: "2026-06-16" + supervisor_schema_version: "2026-06-23" source: "main.go" dags: concurrent_xcom_dag: diff --git a/go-sdk/cmd/airflow-go-pack/pack_test.go b/go-sdk/cmd/airflow-go-pack/pack_test.go index 999d5c0f01771..b11521c4ecaa0 100644 --- a/go-sdk/cmd/airflow-go-pack/pack_test.go +++ b/go-sdk/cmd/airflow-go-pack/pack_test.go @@ -39,7 +39,7 @@ func TestRenderManifest_DeterministicDagOrdering(t *testing.T) { SDK: airflowmetadata.SDK{ Language: "go", Version: "0.1.0", - SupervisorSchemaVersion: "2026-06-16", + SupervisorSchemaVersion: "2026-06-23", }, Dags: map[string]airflowmetadata.Dag{ "zeta_dag": {Tasks: []string{"a", "b"}}, @@ -58,7 +58,7 @@ func TestRenderManifest_DeterministicDagOrdering(t *testing.T) { sdk: language: "go" version: "0.1.0" - supervisor_schema_version: "2026-06-16" + supervisor_schema_version: "2026-06-23" source: "main.go" dags: alpha_dag: @@ -80,7 +80,7 @@ func TestRenderManifest_QuotesValuesNotKeys(t *testing.T) { SDK: airflowmetadata.SDK{ Language: "go", Version: "0.1.0", - SupervisorSchemaVersion: "2026-06-16", + SupervisorSchemaVersion: "2026-06-23", }, Dags: map[string]airflowmetadata.Dag{ "my_dag": {Tasks: []string{"123", "true"}}, @@ -104,7 +104,7 @@ func TestRenderManifest_EmptyDags(t *testing.T) { SDK: airflowmetadata.SDK{ Language: "go", Version: "0.1.0", - SupervisorSchemaVersion: "2026-06-16", + SupervisorSchemaVersion: "2026-06-23", }, Dags: map[string]airflowmetadata.Dag{}, } @@ -235,7 +235,7 @@ func TestRunPack_RejectsOutputAliasingMetadataFile(t *testing.T) { meta := filepath.Join(dir, "airflow-metadata.json") original := []byte( `{"airflow_bundle_metadata_version":"1.0",` + - `"sdk":{"language":"go","version":"0.1.0","supervisor_schema_version":"2026-06-16"},` + + `"sdk":{"language":"go","version":"0.1.0","supervisor_schema_version":"2026-06-23"},` + `"dags":{"my_dag":{"tasks":["t1"]}}}`, ) require.NoError(t, os.WriteFile(meta, original, 0o644)) @@ -346,7 +346,7 @@ func TestRunPack_UsesMetadataFile(t *testing.T) { meta := filepath.Join(dir, "airflow-metadata.json") require.NoError(t, os.WriteFile(meta, []byte( `{"airflow_bundle_metadata_version":"1.0",`+ - `"sdk":{"language":"go","version":"0.1.0","supervisor_schema_version":"2026-06-16"},`+ + `"sdk":{"language":"go","version":"0.1.0","supervisor_schema_version":"2026-06-23"},`+ `"dags":{"my_dag":{"tasks":["t1"]}}}`, ), 0o644)) out := filepath.Join(dir, "bundle") @@ -391,7 +391,7 @@ func TestRunPack_AcceptsYAMLMetadataFile(t *testing.T) { "sdk:\n"+ " language: \"go\"\n"+ " version: \"0.1.0\"\n"+ - " supervisor_schema_version: \"2026-06-16\"\n"+ + " supervisor_schema_version: \"2026-06-23\"\n"+ "dags:\n"+ " yaml_dag:\n"+ " tasks:\n"+ diff --git a/go-sdk/pkg/execution/genmodels/discriminators.gen.go b/go-sdk/pkg/execution/genmodels/discriminators.gen.go index 027a512a7283f..3aff78b9baea1 100644 --- a/go-sdk/pkg/execution/genmodels/discriminators.gen.go +++ b/go-sdk/pkg/execution/genmodels/discriminators.gen.go @@ -24,95 +24,96 @@ import "reflect" // in the supervisor wire-schema: the single source of truth for the "type" // field's value, so callers never hand-write strings that could drift. const ( - TypeAssetEventsResult = "AssetEventsResult" - TypeAssetResult = "AssetResult" - TypeAssetStateStoreResult = "AssetStateStoreResult" - TypeAssetsByAliasResult = "AssetsByAliasResult" - TypeAwaitInputTask = "AwaitInputTask" - TypeClearAssetStateStoreByName = "ClearAssetStateStoreByName" - TypeClearAssetStateStoreByURI = "ClearAssetStateStoreByUri" - TypeClearTaskStateStore = "ClearTaskStateStore" - TypeConnectionResult = "ConnectionResult" - TypeCreateHITLDetailPayload = "CreateHITLDetailPayload" - TypeDRCount = "DRCount" - TypeDagCallbackRequest = "DagCallbackRequest" - TypeDagFileParseRequest = "DagFileParseRequest" - TypeDagFileParsingResult = "DagFileParsingResult" - TypeDagResult = "DagResult" - TypeDagRunResult = "DagRunResult" - TypeDagRunStateResult = "DagRunStateResult" - TypeDeferTask = "DeferTask" - TypeDeleteAssetStateStoreByName = "DeleteAssetStateStoreByName" - TypeDeleteAssetStateStoreByURI = "DeleteAssetStateStoreByUri" - TypeDeleteTaskStateStore = "DeleteTaskStateStore" - TypeDeleteVariable = "DeleteVariable" - TypeDeleteXCom = "DeleteXCom" - TypeEmailRequest = "EmailRequest" - TypeErrorResponse = "ErrorResponse" - TypeGetAssetByName = "GetAssetByName" - TypeGetAssetByURI = "GetAssetByUri" - TypeGetAssetEventByAsset = "GetAssetEventByAsset" - TypeGetAssetEventByAssetAlias = "GetAssetEventByAssetAlias" - TypeGetAssetStateStoreByName = "GetAssetStateStoreByName" - TypeGetAssetStateStoreByURI = "GetAssetStateStoreByUri" - TypeGetAssetsByAlias = "GetAssetsByAlias" - TypeGetConnection = "GetConnection" - TypeGetDRCount = "GetDRCount" - TypeGetDag = "GetDag" - TypeGetDagRun = "GetDagRun" - TypeGetDagRunState = "GetDagRunState" - TypeGetHITLDetailResponse = "GetHITLDetailResponse" - TypeGetPrevSuccessfulDagRun = "GetPrevSuccessfulDagRun" - TypeGetPreviousDagRun = "GetPreviousDagRun" - TypeGetPreviousTI = "GetPreviousTI" - TypeGetTICount = "GetTICount" - TypeGetTaskBreadcrumbs = "GetTaskBreadcrumbs" - TypeGetTaskRescheduleStartDate = "GetTaskRescheduleStartDate" - TypeGetTaskStateStore = "GetTaskStateStore" - TypeGetTaskStates = "GetTaskStates" - TypeGetVariable = "GetVariable" - TypeGetVariableKeys = "GetVariableKeys" - TypeGetXCom = "GetXCom" - TypeGetXComCount = "GetXComCount" - TypeGetXComSequenceItem = "GetXComSequenceItem" - TypeGetXComSequenceSlice = "GetXComSequenceSlice" - TypeHITLDetailRequestResult = "HITLDetailRequestResult" - TypeInactiveAssetsResult = "InactiveAssetsResult" - TypeMaskSecret = "MaskSecret" - TypeOKResponse = "OKResponse" - TypePrevSuccessfulDagRunResult = "PrevSuccessfulDagRunResult" - TypePreviousDagRunResult = "PreviousDagRunResult" - TypePreviousTIResult = "PreviousTIResult" - TypePutVariable = "PutVariable" - TypeRescheduleTask = "RescheduleTask" - TypeResendLoggingFD = "ResendLoggingFD" - TypeRetryTask = "RetryTask" - TypeSentFDs = "SentFDs" - TypeSetAssetStateStoreByName = "SetAssetStateStoreByName" - TypeSetAssetStateStoreByURI = "SetAssetStateStoreByUri" - TypeSetRenderedFields = "SetRenderedFields" - TypeSetRenderedMapIndex = "SetRenderedMapIndex" - TypeSetTaskStateStore = "SetTaskStateStore" - TypeSetXCom = "SetXCom" - TypeSkipDownstreamTasks = "SkipDownstreamTasks" - TypeStartupDetails = "StartupDetails" - TypeSucceedTask = "SucceedTask" - TypeTICount = "TICount" - TypeTaskBreadcrumbsResult = "TaskBreadcrumbsResult" - TypeTaskCallbackRequest = "TaskCallbackRequest" - TypeTaskRescheduleStartDate = "TaskRescheduleStartDate" - TypeTaskState = "TaskState" - TypeTaskStateStoreResult = "TaskStateStoreResult" - TypeTaskStatesResult = "TaskStatesResult" - TypeTriggerDagRun = "TriggerDagRun" - TypeUpdateHITLDetail = "UpdateHITLDetail" - TypeValidateInletsAndOutlets = "ValidateInletsAndOutlets" - TypeVariableKeysResult = "VariableKeysResult" - TypeVariableResult = "VariableResult" - TypeXComCountResponse = "XComCountResponse" - TypeXComResult = "XComResult" - TypeXComSequenceIndexResult = "XComSequenceIndexResult" - TypeXComSequenceSliceResult = "XComSequenceSliceResult" + TypeAssetEventsResult = "AssetEventsResult" + TypeAssetResult = "AssetResult" + TypeAssetStateStoreResult = "AssetStateStoreResult" + TypeAssetsByAliasResult = "AssetsByAliasResult" + TypeAwaitInputTask = "AwaitInputTask" + TypeClearAssetStateStoreByName = "ClearAssetStateStoreByName" + TypeClearAssetStateStoreByURI = "ClearAssetStateStoreByUri" + TypeClearTaskStateStore = "ClearTaskStateStore" + TypeConnectionResult = "ConnectionResult" + TypeCreateHITLDetailPayload = "CreateHITLDetailPayload" + TypeDRCount = "DRCount" + TypeDagCallbackRequest = "DagCallbackRequest" + TypeDagFileParseRequest = "DagFileParseRequest" + TypeDagFileParsingResult = "DagFileParsingResult" + TypeDagResult = "DagResult" + TypeDagRunResult = "DagRunResult" + TypeDagRunStateResult = "DagRunStateResult" + TypeDagSkippedIntervalsCallbackRequest = "DagSkippedIntervalsCallbackRequest" + TypeDeferTask = "DeferTask" + TypeDeleteAssetStateStoreByName = "DeleteAssetStateStoreByName" + TypeDeleteAssetStateStoreByURI = "DeleteAssetStateStoreByUri" + TypeDeleteTaskStateStore = "DeleteTaskStateStore" + TypeDeleteVariable = "DeleteVariable" + TypeDeleteXCom = "DeleteXCom" + TypeEmailRequest = "EmailRequest" + TypeErrorResponse = "ErrorResponse" + TypeGetAssetByName = "GetAssetByName" + TypeGetAssetByURI = "GetAssetByUri" + TypeGetAssetEventByAsset = "GetAssetEventByAsset" + TypeGetAssetEventByAssetAlias = "GetAssetEventByAssetAlias" + TypeGetAssetStateStoreByName = "GetAssetStateStoreByName" + TypeGetAssetStateStoreByURI = "GetAssetStateStoreByUri" + TypeGetAssetsByAlias = "GetAssetsByAlias" + TypeGetConnection = "GetConnection" + TypeGetDRCount = "GetDRCount" + TypeGetDag = "GetDag" + TypeGetDagRun = "GetDagRun" + TypeGetDagRunState = "GetDagRunState" + TypeGetHITLDetailResponse = "GetHITLDetailResponse" + TypeGetPrevSuccessfulDagRun = "GetPrevSuccessfulDagRun" + TypeGetPreviousDagRun = "GetPreviousDagRun" + TypeGetPreviousTI = "GetPreviousTI" + TypeGetTICount = "GetTICount" + TypeGetTaskBreadcrumbs = "GetTaskBreadcrumbs" + TypeGetTaskRescheduleStartDate = "GetTaskRescheduleStartDate" + TypeGetTaskStateStore = "GetTaskStateStore" + TypeGetTaskStates = "GetTaskStates" + TypeGetVariable = "GetVariable" + TypeGetVariableKeys = "GetVariableKeys" + TypeGetXCom = "GetXCom" + TypeGetXComCount = "GetXComCount" + TypeGetXComSequenceItem = "GetXComSequenceItem" + TypeGetXComSequenceSlice = "GetXComSequenceSlice" + TypeHITLDetailRequestResult = "HITLDetailRequestResult" + TypeInactiveAssetsResult = "InactiveAssetsResult" + TypeMaskSecret = "MaskSecret" + TypeOKResponse = "OKResponse" + TypePrevSuccessfulDagRunResult = "PrevSuccessfulDagRunResult" + TypePreviousDagRunResult = "PreviousDagRunResult" + TypePreviousTIResult = "PreviousTIResult" + TypePutVariable = "PutVariable" + TypeRescheduleTask = "RescheduleTask" + TypeResendLoggingFD = "ResendLoggingFD" + TypeRetryTask = "RetryTask" + TypeSentFDs = "SentFDs" + TypeSetAssetStateStoreByName = "SetAssetStateStoreByName" + TypeSetAssetStateStoreByURI = "SetAssetStateStoreByUri" + TypeSetRenderedFields = "SetRenderedFields" + TypeSetRenderedMapIndex = "SetRenderedMapIndex" + TypeSetTaskStateStore = "SetTaskStateStore" + TypeSetXCom = "SetXCom" + TypeSkipDownstreamTasks = "SkipDownstreamTasks" + TypeStartupDetails = "StartupDetails" + TypeSucceedTask = "SucceedTask" + TypeTICount = "TICount" + TypeTaskBreadcrumbsResult = "TaskBreadcrumbsResult" + TypeTaskCallbackRequest = "TaskCallbackRequest" + TypeTaskRescheduleStartDate = "TaskRescheduleStartDate" + TypeTaskState = "TaskState" + TypeTaskStateStoreResult = "TaskStateStoreResult" + TypeTaskStatesResult = "TaskStatesResult" + TypeTriggerDagRun = "TriggerDagRun" + TypeUpdateHITLDetail = "UpdateHITLDetail" + TypeValidateInletsAndOutlets = "ValidateInletsAndOutlets" + TypeVariableKeysResult = "VariableKeysResult" + TypeVariableResult = "VariableResult" + TypeXComCountResponse = "XComCountResponse" + TypeXComResult = "XComResult" + TypeXComSequenceIndexResult = "XComSequenceIndexResult" + TypeXComSequenceSliceResult = "XComSequenceSliceResult" ) // EnsureType returns m with its "type" discriminator set to the constant bound to @@ -172,6 +173,9 @@ func EnsureType(m any) any { case DagRunStateResult: b.Type = TypeDagRunStateResult return b + case DagSkippedIntervalsCallbackRequest: + b.Type = TypeDagSkippedIntervalsCallbackRequest + return b case DeferTask: b.Type = TypeDeferTask return b diff --git a/go-sdk/pkg/execution/genmodels/models.gen.go b/go-sdk/pkg/execution/genmodels/models.gen.go index e6861d8c8add5..f55d44b776d14 100644 --- a/go-sdk/pkg/execution/genmodels/models.gen.go +++ b/go-sdk/pkg/execution/genmodels/models.gen.go @@ -370,6 +370,9 @@ type DagCallbackRequest struct { // Type corresponds to the JSON schema field "type". Type string `msgpack:"type,omitempty"` + + // VersionData corresponds to the JSON schema field "version_data". + VersionData *VersionData `msgpack:"version_data,omitempty"` } // Request for DAG File Parsing. @@ -627,6 +630,33 @@ const DagRunTypeManual DagRunType = "manual" const DagRunTypeOperatorTriggered DagRunType = "operator_triggered" const DagRunTypeScheduled DagRunType = "scheduled" +// Store skipped intervals callback data for execution by the Dag processor. +type DagSkippedIntervalsCallbackRequest struct { + // BundleName corresponds to the JSON schema field "bundle_name". + BundleName string `msgpack:"bundle_name"` + + // BundleVersion corresponds to the JSON schema field "bundle_version". + BundleVersion interface{} `msgpack:"bundle_version"` + + // DagID corresponds to the JSON schema field "dag_id". + DagID string `msgpack:"dag_id"` + + // Filepath corresponds to the JSON schema field "filepath". + Filepath string `msgpack:"filepath"` + + // Msg corresponds to the JSON schema field "msg". + Msg interface{} `msgpack:"msg,omitempty"` + + // SkippedRange corresponds to the JSON schema field "skipped_range". + SkippedRange []interface{} `msgpack:"skipped_range"` + + // Type corresponds to the JSON schema field "type". + Type string `msgpack:"type,omitempty"` + + // VersionData corresponds to the JSON schema field "version_data". + VersionData *VersionData `msgpack:"version_data,omitempty"` +} + type Data map[string]interface{} type Defaults []string @@ -749,6 +779,9 @@ type EmailRequest struct { // Type corresponds to the JSON schema field "type". Type string `msgpack:"type,omitempty"` + + // VersionData corresponds to the JSON schema field "version_data". + VersionData *VersionData `msgpack:"version_data,omitempty"` } type EmailRequestEmailType string @@ -808,6 +841,9 @@ type GetAssetEventByAsset struct { // Before corresponds to the JSON schema field "before". Before interface{} `msgpack:"before,omitempty"` + // Extra corresponds to the JSON schema field "extra". + Extra *Extra `msgpack:"extra,omitempty"` + // Limit corresponds to the JSON schema field "limit". Limit interface{} `msgpack:"limit,omitempty"` @@ -834,6 +870,9 @@ type GetAssetEventByAssetAlias struct { // Before corresponds to the JSON schema field "before". Before interface{} `msgpack:"before,omitempty"` + // Extra corresponds to the JSON schema field "extra". + Extra *Extra `msgpack:"extra,omitempty"` + // Limit corresponds to the JSON schema field "limit". Limit interface{} `msgpack:"limit,omitempty"` @@ -1606,36 +1645,17 @@ type TaskBreadcrumbsResult struct { type TaskBreadcrumbsResultBreadcrumbsElem map[string]interface{} -// Task callback status information. -// -// A Class with information about the success/failure TI callback to be executed. -// Currently, only failure -// callbacks when tasks are externally killed or experience heartbeat timeouts are -// run via DagFileProcessorProcess. -type TaskCallbackRequest struct { - // BundleName corresponds to the JSON schema field "bundle_name". - BundleName string `msgpack:"bundle_name"` - - // BundleVersion corresponds to the JSON schema field "bundle_version". - BundleVersion interface{} `msgpack:"bundle_version"` - - // ContextFromServer corresponds to the JSON schema field "context_from_server". - ContextFromServer *TIRunContext `msgpack:"context_from_server,omitempty"` - - // Filepath corresponds to the JSON schema field "filepath". - Filepath string `msgpack:"filepath"` - - // Msg corresponds to the JSON schema field "msg". - Msg interface{} `msgpack:"msg,omitempty"` +type Warnings []interface{} - // TaskCallbackType corresponds to the JSON schema field "task_callback_type". - TaskCallbackType interface{} `msgpack:"task_callback_type,omitempty"` +type TriggerKwargs map[string]JsonValue - // TI corresponds to the JSON schema field "ti". - TI TaskInstance `msgpack:"ti"` +// Variable schema for responses with fields that are needed for Runtime. +type VariableResponse struct { + // Key corresponds to the JSON schema field "key". + Key string `msgpack:"key"` - // Type corresponds to the JSON schema field "type". - Type string `msgpack:"type,omitempty"` + // Value corresponds to the JSON schema field "value". + Value interface{} `msgpack:"value"` } type TaskIds []string @@ -1673,24 +1693,60 @@ type TaskInstance struct { TryNumber int `msgpack:"try_number"` } -type TaskInstanceState string - const TaskInstanceStateAwaitingInput TaskInstanceState = "awaiting_input" const TaskInstanceStateDeferred TaskInstanceState = "deferred" +const TaskInstanceStateSkipped TaskInstanceState = "skipped" +const TaskInstanceStateUpstreamFailed TaskInstanceState = "upstream_failed" +const TaskInstanceStateUpForReschedule TaskInstanceState = "up_for_reschedule" +const TaskInstanceStateUpForRetry TaskInstanceState = "up_for_retry" const TaskInstanceStateFailed TaskInstanceState = "failed" -const TaskInstanceStateQueued TaskInstanceState = "queued" -const TaskInstanceStateRemoved TaskInstanceState = "removed" const TaskInstanceStateRestarting TaskInstanceState = "restarting" +const TaskInstanceStateSuccess TaskInstanceState = "success" const TaskInstanceStateRunning TaskInstanceState = "running" +const TaskInstanceStateQueued TaskInstanceState = "queued" const TaskInstanceStateScheduled TaskInstanceState = "scheduled" -const TaskInstanceStateSkipped TaskInstanceState = "skipped" -const TaskInstanceStateSuccess TaskInstanceState = "success" -const TaskInstanceStateUpForReschedule TaskInstanceState = "up_for_reschedule" -const TaskInstanceStateUpForRetry TaskInstanceState = "up_for_retry" -const TaskInstanceStateUpstreamFailed TaskInstanceState = "upstream_failed" type TaskOutlets []AssetProfile +const TaskInstanceStateRemoved TaskInstanceState = "removed" + +type TaskInstanceState string + +// Task callback status information. +// +// A Class with information about the success/failure TI callback to be executed. +// Currently, only failure +// callbacks when tasks are externally killed or experience heartbeat timeouts are +// run via DagFileProcessorProcess. +type TaskCallbackRequest struct { + // BundleName corresponds to the JSON schema field "bundle_name". + BundleName string `msgpack:"bundle_name"` + + // BundleVersion corresponds to the JSON schema field "bundle_version". + BundleVersion interface{} `msgpack:"bundle_version"` + + // ContextFromServer corresponds to the JSON schema field "context_from_server". + ContextFromServer *TIRunContext `msgpack:"context_from_server,omitempty"` + + // Filepath corresponds to the JSON schema field "filepath". + Filepath string `msgpack:"filepath"` + + // Msg corresponds to the JSON schema field "msg". + Msg interface{} `msgpack:"msg,omitempty"` + + // TaskCallbackType corresponds to the JSON schema field "task_callback_type". + TaskCallbackType interface{} `msgpack:"task_callback_type,omitempty"` + + // TI corresponds to the JSON schema field "ti". + TI TaskInstance `msgpack:"ti"` + + // Type corresponds to the JSON schema field "type". + Type string `msgpack:"type,omitempty"` + + // VersionData corresponds to the JSON schema field "version_data". + VersionData *VersionData `msgpack:"version_data,omitempty"` +} + // Response containing the first reschedule date for a task instance. type TaskRescheduleStartDate struct { // StartDate corresponds to the JSON schema field "start_date". @@ -1700,6 +1756,12 @@ type TaskRescheduleStartDate struct { Type string `msgpack:"type,omitempty"` } +type TaskStateState string + +const TaskStateStateFailed TaskStateState = "failed" +const TaskStateStateSkipped TaskStateState = "skipped" +const TaskStateStateRemoved TaskStateState = "removed" + // Update a task's state. // // If a process exits without sending one of these the state will be derived from @@ -1720,12 +1782,6 @@ type TaskState struct { Type string `msgpack:"type,omitempty"` } -type TaskStateState string - -const TaskStateStateFailed TaskStateState = "failed" -const TaskStateStateRemoved TaskStateState = "removed" -const TaskStateStateSkipped TaskStateState = "skipped" - // Response to GetTaskStateStore; wraps the generated API response for supervisor // to worker comms. type TaskStateStoreResult struct { @@ -1775,19 +1831,6 @@ type TriggerDagRun struct { Type string `msgpack:"type,omitempty"` } -type TriggerKwargs map[string]JsonValue - -type Warnings []interface{} - -// Variable schema for responses with fields that are needed for Runtime. -type VariableResponse struct { - // Key corresponds to the JSON schema field "key". - Key string `msgpack:"key"` - - // Value corresponds to the JSON schema field "value". - Value interface{} `msgpack:"value"` -} - type VersionData map[string]interface{} // Update the response content part of an existing Human-in-the-loop response. diff --git a/go-sdk/pkg/execution/messages.go b/go-sdk/pkg/execution/messages.go index 72d451a866632..0759b6bd3d42f 100644 --- a/go-sdk/pkg/execution/messages.go +++ b/go-sdk/pkg/execution/messages.go @@ -32,7 +32,7 @@ import ( // reported in a bundle's airflow-metadata manifest as // sdk.supervisor_schema_version so the supervisor can down/upgrade messages to // a shape the bundle understands. -const SupervisorSchemaVersion = "2026-06-16" +const SupervisorSchemaVersion = "2026-06-23" // The message-type discriminator strings (genmodels.Type*) are generated from the // schema's "type" consts in discriminators.gen.go; outbound messages stamp the diff --git a/go-sdk/pkg/execution/metadata_test.go b/go-sdk/pkg/execution/metadata_test.go index d105b24d898c3..d9543b700eb99 100644 --- a/go-sdk/pkg/execution/metadata_test.go +++ b/go-sdk/pkg/execution/metadata_test.go @@ -35,7 +35,7 @@ func sampleManifest() airflowmetadata.Manifest { SDK: airflowmetadata.SDK{ Language: "go", Version: "(devel)", - SupervisorSchemaVersion: "2026-06-16", + SupervisorSchemaVersion: "2026-06-23", }, Dags: map[string]airflowmetadata.Dag{ "simple_dag": {Tasks: []string{"extract", "transform", "load"}}, diff --git a/scripts/in_container/run_schema_defaults_check.py b/scripts/in_container/run_schema_defaults_check.py index 7afcbd9ce546f..cc29526cb2983 100755 --- a/scripts/in_container/run_schema_defaults_check.py +++ b/scripts/in_container/run_schema_defaults_check.py @@ -230,6 +230,7 @@ def compare_dag_defaults() -> list[str]: computed_properties = { "has_on_success_callback", "has_on_failure_callback", + "has_on_skipped_intervals_callback", } if field_name not in computed_properties: errors.append( diff --git a/task-sdk/docs/api.rst b/task-sdk/docs/api.rst index 51aa275b90bb0..fef72eb2f5bad 100644 --- a/task-sdk/docs/api.rst +++ b/task-sdk/docs/api.rst @@ -315,6 +315,10 @@ For a complete list of available context variables (such as ``dag_run``, ``task_instance``, ``logical_date``, etc.), see the :ref:`Templates reference `. +.. autoclass:: airflow.sdk.SkippedIntervalsCallbackContext + +The ``SkippedIntervalsCallbackContext`` object is passed to ``on_skipped_intervals_callback`` handlers. + .. rubric:: Task State .. autodata:: airflow.sdk.NEVER_EXPIRE diff --git a/task-sdk/docs/index.rst b/task-sdk/docs/index.rst index 18e215d4ce7d7..3face7bb88d3e 100644 --- a/task-sdk/docs/index.rst +++ b/task-sdk/docs/index.rst @@ -96,6 +96,7 @@ Why use ``airflow.sdk``? - :class:`airflow.sdk.RetryDecision` - :class:`airflow.sdk.RetryPolicy` - :class:`airflow.sdk.RetryRule` +- :class:`airflow.sdk.SkippedIntervalsCallbackContext` - :class:`airflow.sdk.TaskGroup` - :class:`airflow.sdk.TaskInstanceState` - :class:`airflow.sdk.DagRunState` diff --git a/task-sdk/src/airflow/sdk/__init__.py b/task-sdk/src/airflow/sdk/__init__.py index 807019eade80e..cc386a2399a23 100644 --- a/task-sdk/src/airflow/sdk/__init__.py +++ b/task-sdk/src/airflow/sdk/__init__.py @@ -81,6 +81,7 @@ "RollupMapper", "SegmentWindow", "SkipMixin", + "SkippedIntervalsCallbackContext", "SyncCallback", "StartOfDayMapper", "StartOfHourMapper", @@ -149,7 +150,12 @@ from airflow.sdk.definitions.asset.metadata import Metadata from airflow.sdk.definitions.callback import AsyncCallback, SyncCallback from airflow.sdk.definitions.connection import Connection - from airflow.sdk.definitions.context import Context, get_current_context, get_parsing_context + from airflow.sdk.definitions.context import ( + Context, + SkippedIntervalsCallbackContext, + get_current_context, + get_parsing_context, + ) from airflow.sdk.definitions.dag import DAG, dag from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference from airflow.sdk.definitions.decorators import result, setup, task, teardown @@ -284,6 +290,7 @@ "SecretCache": ".execution_time.cache", "SegmentWindow": ".definitions.partition_mappers.window", "SkipMixin": ".bases.skipmixin", + "SkippedIntervalsCallbackContext": ".definitions.context", "SyncCallback": ".definitions.callback", "StartOfDayMapper": ".definitions.partition_mappers.temporal", "StartOfHourMapper": ".definitions.partition_mappers.temporal", diff --git a/task-sdk/src/airflow/sdk/definitions/context.py b/task-sdk/src/airflow/sdk/definitions/context.py index 9d36203c2ebbf..e713c9c6d50c8 100644 --- a/task-sdk/src/airflow/sdk/definitions/context.py +++ b/task-sdk/src/airflow/sdk/definitions/context.py @@ -20,7 +20,7 @@ import copy import os from collections.abc import MutableMapping -from typing import TYPE_CHECKING, Any, NamedTuple, TypedDict, cast +from typing import TYPE_CHECKING, Any, Literal, NamedTuple, TypedDict, cast from typing_extensions import NotRequired @@ -41,6 +41,7 @@ OutletEventAccessorsProtocol, RuntimeTaskInstanceProtocol, ) + from airflow.timetables.base import DataInterval # noqa: SDK002 class Context(TypedDict, total=False): @@ -92,6 +93,14 @@ class Context(TypedDict, total=False): var: Any +class SkippedIntervalsCallbackContext(TypedDict): + """Context passed to ``on_skipped_intervals_callback`` handlers.""" + + dag: DAG + reason: Literal["skipped_intervals"] + skipped_range: DataInterval + + KNOWN_CONTEXT_KEYS: set[str] = set(Context.__annotations__.keys()) diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py b/task-sdk/src/airflow/sdk/definitions/dag.py index cadfe6978300e..fd8967d358fa5 100644 --- a/task-sdk/src/airflow/sdk/definitions/dag.py +++ b/task-sdk/src/airflow/sdk/definitions/dag.py @@ -46,7 +46,7 @@ from airflow.sdk.definitions._internal.node import DAGNode, validate_key from airflow.sdk.definitions._internal.types import NOTSET, ArgNotSet, is_arg_set from airflow.sdk.definitions.asset import AssetAll, BaseAsset -from airflow.sdk.definitions.context import Context +from airflow.sdk.definitions.context import Context, SkippedIntervalsCallbackContext from airflow.sdk.definitions.deadline import DeadlineAlert from airflow.sdk.definitions.param import DagParam, ParamsDict from airflow.sdk.definitions.timetables.assets import AssetTriggeredTimetable @@ -102,6 +102,7 @@ ) DagStateChangeCallback = Callable[[Context], None] +SkippedIntervalsCallback = Callable[[SkippedIntervalsCallbackContext], None] ScheduleInterval = None | str | timedelta | relativedelta ScheduleArg = Union[ScheduleInterval, BaseTimetable, "CoreTimetable", BaseAsset, Collection[BaseAsset]] @@ -388,6 +389,12 @@ class DAG: A context dictionary is passed as a single parameter to this function. :param on_success_callback: Much like the ``on_failure_callback`` except that it is executed when the dag succeeds. + :param on_skipped_intervals_callback: A function or list of functions invoked by the + scheduler when a Dag with ``catchup=False`` advances past one or more scheduled + data intervals without creating Dag runs for them (for example after a scheduler + restart or when a paused Dag is re-enabled). The callback receives a + :class:`~airflow.sdk.definitions.context.SkippedIntervalsCallbackContext`. + It runs in the dag processor, not in the scheduler. :param access_control: Specify optional DAG-level actions, e.g., "{'role1': {'can_read'}, 'role2': {'can_read', 'can_edit', 'can_delete'}}" or it can specify the resource name if there is a DAGs Run resource, e.g., @@ -517,6 +524,7 @@ def __rich_repr__(self): ) on_success_callback: None | DagStateChangeCallback | list[DagStateChangeCallback] = None on_failure_callback: None | DagStateChangeCallback | list[DagStateChangeCallback] = None + on_skipped_intervals_callback: None | SkippedIntervalsCallback | list[SkippedIntervalsCallback] = None doc_md: str | None = attrs.field(default=None, converter=_convert_doc_md) params: ParamsDict = attrs.field( # mypy doesn't really like passing the Converter object @@ -558,6 +566,7 @@ def __rich_repr__(self): has_on_success_callback: bool = attrs.field(init=False) has_on_failure_callback: bool = attrs.field(init=False) + has_on_skipped_intervals_callback: bool = attrs.field(init=False) disable_bundle_versioning: bool = attrs.field( factory=_config_bool_factory("dag_processor", "disable_bundle_versioning") ) @@ -706,6 +715,10 @@ def _has_on_success_callback(self) -> bool: def _has_on_failure_callback(self) -> bool: return self.on_failure_callback is not None + @has_on_skipped_intervals_callback.default + def _has_on_skipped_intervals_callback(self) -> bool: + return self.on_skipped_intervals_callback is not None + @sla_miss_callback.validator def _validate_sla_miss_callback(self, _, value): if value is not None: @@ -1599,11 +1612,13 @@ def _run_inline_trigger(trigger, task_sdk_ti): "sla_miss_callback", "on_success_callback", "on_failure_callback", + "on_skipped_intervals_callback", "template_undefined", "jinja_environment_kwargs", # has_on_*_callback are only stored if the value is True, as the default is False "has_on_success_callback", "has_on_failure_callback", + "has_on_skipped_intervals_callback", "auto_register", "schedule", } @@ -1631,6 +1646,9 @@ def dag( catchup: bool = ..., on_success_callback: None | DagStateChangeCallback | list[DagStateChangeCallback] = None, on_failure_callback: None | DagStateChangeCallback | list[DagStateChangeCallback] = None, + on_skipped_intervals_callback: None + | SkippedIntervalsCallback + | list[SkippedIntervalsCallback] = None, deadline: list[DeadlineAlert] | DeadlineAlert | None = None, doc_md: str | None = None, params: ParamsDict | dict[str, Any] | None = None, diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json index 0ec8fe4e49a1c..f820cbc0d72c8 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json +++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json @@ -1,6 +1,6 @@ { "$schema": "https://json-schema.org/draft/2020-12/schema", - "api_version": "2026-06-16", + "api_version": "2026-06-23", "description": "Apache Airflow SDK Supervisor Schema", "$defs": { "AssetAliasReferenceAssetEventDagRun": { @@ -847,6 +847,7 @@ "discriminator": { "mapping": { "DagCallbackRequest": "#/$defs/DagCallbackRequest", + "DagSkippedIntervalsCallbackRequest": "#/$defs/DagSkippedIntervalsCallbackRequest", "EmailRequest": "#/$defs/EmailRequest", "TaskCallbackRequest": "#/$defs/TaskCallbackRequest" }, @@ -856,6 +857,9 @@ { "$ref": "#/$defs/DagCallbackRequest" }, + { + "$ref": "#/$defs/DagSkippedIntervalsCallbackRequest" + }, { "$ref": "#/$defs/TaskCallbackRequest" }, @@ -1395,6 +1399,90 @@ "title": "DagRunType", "type": "string" }, + "DagSkippedIntervalsCallbackRequest": { + "description": "Store skipped intervals callback data for execution by the Dag processor.", + "properties": { + "filepath": { + "title": "Filepath", + "type": "string" + }, + "bundle_name": { + "title": "Bundle Name", + "type": "string" + }, + "bundle_version": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Bundle Version" + }, + "version_data": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Version Data" + }, + "msg": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Msg" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "skipped_range": { + "maxItems": 2, + "minItems": 2, + "prefixItems": [ + { + "format": "date-time", + "type": "string" + }, + { + "format": "date-time", + "type": "string" + } + ], + "title": "Skipped Range", + "type": "array" + }, + "type": { + "const": "DagSkippedIntervalsCallbackRequest", + "default": "DagSkippedIntervalsCallbackRequest", + "title": "Type", + "type": "string" + } + }, + "required": [ + "filepath", + "bundle_name", + "bundle_version", + "dag_id", + "skipped_range" + ], + "title": "DagSkippedIntervalsCallbackRequest", + "type": "object" + }, "DeferTask": { "additionalProperties": false, "description": "Update a task instance state to deferred.", diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py index 9491a8993fdc3..3a8af44cf270f 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py +++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py @@ -23,6 +23,8 @@ if TYPE_CHECKING: from cadwyn import VersionBundle +from airflow.sdk.execution_time.schema.versions.v2026_06_23 import AddDagSkippedIntervalsCallbackRequest + @functools.cache def get_bundle() -> VersionBundle: @@ -39,6 +41,7 @@ def get_bundle() -> VersionBundle: return VersionBundle( HeadVersion(), + Version("2026-06-23", AddDagSkippedIntervalsCallbackRequest), Version("2026-06-16"), ) diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_06_23.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_06_23.py new file mode 100644 index 0000000000000..ff46209d375c2 --- /dev/null +++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_06_23.py @@ -0,0 +1,55 @@ +# 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 logging + +from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for + +from airflow.dag_processing.processor import DagFileParseRequest # noqa: SDK002 + +logger = logging.getLogger(__name__) + +_SKIPPED_INTERVALS_CALLBACK_TYPE = "DagSkippedIntervalsCallbackRequest" + + +class AddDagSkippedIntervalsCallbackRequest(VersionChange): + """Introduce ``DagSkippedIntervalsCallbackRequest`` in the ``CallbackRequest`` union.""" + + description = __doc__ + + instructions_to_migrate_to_previous_version = () + + @convert_response_to_previous_version_for(DagFileParseRequest) # type: ignore[arg-type] + def _drop_dag_skipped_intervals_callbacks(response: ResponseInfo) -> None: # type: ignore[misc] + callbacks = response.body.get("callback_requests") + if not callbacks: + return + + filtered: list[object] = [] + for callback in callbacks: + if isinstance(callback, dict) and callback.get("type") == _SKIPPED_INTERVALS_CALLBACK_TYPE: + logger.info( + "Dropping unsupported callback request for supervisor schema downgrade", + extra={ + "callback_type": _SKIPPED_INTERVALS_CALLBACK_TYPE, + "dag_id": callback.get("dag_id"), + }, + ) + continue + filtered.append(callback) + response.body["callback_requests"] = filtered diff --git a/ts-sdk/src/generated/supervisor.ts b/ts-sdk/src/generated/supervisor.ts index ab2632831ab95..5885ecf5da189 100644 --- a/ts-sdk/src/generated/supervisor.ts +++ b/ts-sdk/src/generated/supervisor.ts @@ -199,6 +199,20 @@ export type VersionData2 = { [k: string]: unknown; } | null; export type Msg1 = string | null; +export type DagId4 = string; +/** + * @minItems 2 + * @maxItems 2 + */ +export type SkippedRange = [unknown, unknown]; +export type Type13 = "DagSkippedIntervalsCallbackRequest"; +export type Filepath2 = string; +export type BundleName3 = string; +export type BundleVersion2 = string | null; +export type VersionData3 = { + [k: string]: unknown; +} | null; +export type Msg2 = string | null; /** * All possible states that a Task Instance can be in. * @@ -245,18 +259,23 @@ export type NextKwargs1 = export type XcomKeysToClear = string[]; export type ShouldRetry = boolean; export type StartDate2 = string | null; -export type Type13 = "TaskCallbackRequest"; -export type Filepath2 = string; -export type BundleName3 = string; -export type BundleVersion2 = string | null; -export type VersionData3 = { +export type Type14 = "TaskCallbackRequest"; +export type Filepath3 = string; +export type BundleName4 = string; +export type BundleVersion3 = string | null; +export type VersionData4 = { [k: string]: unknown; } | null; -export type Msg2 = string | null; +export type Msg3 = string | null; export type EmailType = "failure" | "retry"; -export type Type14 = "EmailRequest"; -export type CallbackRequests = (DagCallbackRequest | TaskCallbackRequest | EmailRequest)[]; -export type Type15 = "DagFileParseRequest"; +export type Type15 = "EmailRequest"; +export type CallbackRequests = ( + | DagCallbackRequest + | DagSkippedIntervalsCallbackRequest + | TaskCallbackRequest + | EmailRequest +)[]; +export type Type16 = "DagFileParseRequest"; export type Fileloc = string; export type LastLoaded = string | null; export type SerializedDags = LazyDeserializedDAG[]; @@ -264,17 +283,17 @@ export type Warnings = unknown[] | null; export type ImportErrors = { [k: string]: string; } | null; -export type Type16 = "DagFileParsingResult"; -export type DagId4 = string; +export type Type17 = "DagFileParsingResult"; +export type DagId5 = string; export type IsPaused = boolean; -export type BundleName4 = string | null; -export type BundleVersion3 = string | null; +export type BundleName5 = string | null; +export type BundleVersion4 = string | null; export type RelativeFileloc = string | null; export type Owners = string | null; export type Tags = string[]; export type NextDagrun = string | null; -export type Type17 = "DagResult"; -export type DagId5 = string; +export type Type18 = "DagResult"; +export type DagId6 = string; export type RunId4 = string; export type LogicalDate2 = string | null; export type DataIntervalStart2 = string | null; @@ -292,8 +311,8 @@ export type PartitionKey4 = string | null; export type PartitionDate1 = string | null; export type Note1 = string | null; export type TeamName1 = string | null; -export type Type18 = "DagRunResult"; -export type Type19 = "DagRunStateResult"; +export type Type19 = "DagRunResult"; +export type Type20 = "DagRunStateResult"; export type State2 = "deferred" | null; export type Classpath = string; export type TriggerKwargs = @@ -309,24 +328,24 @@ export type NextKwargs2 = { [k: string]: JsonValue; } | null; export type RenderedMapIndex1 = string | null; -export type Type20 = "DeferTask"; +export type Type21 = "DeferTask"; export type Name8 = string; export type Key1 = string; -export type Type21 = "DeleteAssetStateStoreByName"; +export type Type22 = "DeleteAssetStateStoreByName"; export type Uri5 = string; export type Key2 = string; -export type Type22 = "DeleteAssetStateStoreByUri"; +export type Type23 = "DeleteAssetStateStoreByUri"; export type TiId2 = string; export type Key3 = string; -export type Type23 = "DeleteTaskStateStore"; +export type Type24 = "DeleteTaskStateStore"; export type Key4 = string; -export type Type24 = "DeleteVariable"; +export type Type25 = "DeleteVariable"; export type Key5 = string; -export type DagId6 = string; +export type DagId7 = string; export type RunId5 = string; export type TaskId1 = string; export type MapIndex1 = number | null; -export type Type25 = "DeleteXCom"; +export type Type26 = "DeleteXCom"; /** * Error types used in the API client. */ @@ -344,7 +363,7 @@ export type ErrorType = export type Detail = { [k: string]: unknown; } | null; -export type Type26 = "ErrorResponse"; +export type Type27 = "ErrorResponse"; /** * Error types used in the API client. * @@ -363,9 +382,9 @@ export type ErrorType1 = | "GENERIC_ERROR" | "API_SERVER_ERROR"; export type Name9 = string; -export type Type27 = "GetAssetByName"; +export type Type28 = "GetAssetByName"; export type Uri6 = string; -export type Type28 = "GetAssetByUri"; +export type Type29 = "GetAssetByUri"; export type Name10 = string | null; export type Uri7 = string | null; export type After = string | null; @@ -377,7 +396,7 @@ export type PartitionKeyRegexpPattern = string | null; export type Extra7 = { [k: string]: string; } | null; -export type Type29 = "GetAssetEventByAsset"; +export type Type30 = "GetAssetEventByAsset"; export type AliasName = string; export type After1 = string | null; export type Before1 = string | null; @@ -388,100 +407,100 @@ export type PartitionKeyRegexpPattern1 = string | null; export type Extra8 = { [k: string]: string; } | null; -export type Type30 = "GetAssetEventByAssetAlias"; +export type Type31 = "GetAssetEventByAssetAlias"; export type Name11 = string; export type Key6 = string; -export type Type31 = "GetAssetStateStoreByName"; +export type Type32 = "GetAssetStateStoreByName"; export type Uri8 = string; export type Key7 = string; -export type Type32 = "GetAssetStateStoreByUri"; +export type Type33 = "GetAssetStateStoreByUri"; export type AliasName1 = string; -export type Type33 = "GetAssetsByAlias"; +export type Type34 = "GetAssetsByAlias"; export type ConnId2 = string; -export type Type34 = "GetConnection"; -export type DagId7 = string; +export type Type35 = "GetConnection"; +export type DagId8 = string; export type LogicalDates = string[] | null; export type RunIds = string[] | null; export type States = string[] | null; -export type Type35 = "GetDRCount"; -export type DagId8 = string; -export type Type36 = "GetDag"; +export type Type36 = "GetDRCount"; export type DagId9 = string; -export type RunId6 = string; -export type Type37 = "GetDagRun"; +export type Type37 = "GetDag"; export type DagId10 = string; +export type RunId6 = string; +export type Type38 = "GetDagRun"; +export type DagId11 = string; export type RunId7 = string; -export type Type38 = "GetDagRunState"; +export type Type39 = "GetDagRunState"; export type TiId3 = string; -export type Type39 = "GetHITLDetailResponse"; +export type Type40 = "GetHITLDetailResponse"; export type TiId4 = string; -export type Type40 = "GetPrevSuccessfulDagRun"; -export type DagId11 = string; +export type Type41 = "GetPrevSuccessfulDagRun"; +export type DagId12 = string; export type LogicalDate3 = string; export type State3 = string | null; -export type Type41 = "GetPreviousDagRun"; -export type DagId12 = string; +export type Type42 = "GetPreviousDagRun"; +export type DagId13 = string; export type TaskId2 = string; export type LogicalDate4 = string | null; export type MapIndex2 = number; -export type Type42 = "GetPreviousTI"; -export type DagId13 = string; +export type Type43 = "GetPreviousTI"; +export type DagId14 = string; export type MapIndex3 = number | null; export type TaskIds = string[] | null; export type TaskGroupId = string | null; export type LogicalDates1 = string[] | null; export type RunIds1 = string[] | null; export type States1 = string[] | null; -export type Type43 = "GetTICount"; -export type DagId14 = string; +export type Type44 = "GetTICount"; +export type DagId15 = string; export type RunId8 = string; -export type Type44 = "GetTaskBreadcrumbs"; +export type Type45 = "GetTaskBreadcrumbs"; export type TiId5 = string; export type TryNumber1 = number; -export type Type45 = "GetTaskRescheduleStartDate"; +export type Type46 = "GetTaskRescheduleStartDate"; export type TiId6 = string; export type Key8 = string; -export type Type46 = "GetTaskStateStore"; -export type DagId15 = string; +export type Type47 = "GetTaskStateStore"; +export type DagId16 = string; export type MapIndex4 = number | null; export type TaskIds1 = string[] | null; export type TaskGroupId1 = string | null; export type LogicalDates2 = string[] | null; export type RunIds2 = string[] | null; -export type Type47 = "GetTaskStates"; +export type Type48 = "GetTaskStates"; export type Key9 = string; -export type Type48 = "GetVariable"; +export type Type49 = "GetVariable"; export type Prefix = string | null; export type Limit2 = number; export type Offset = number; -export type Type49 = "GetVariableKeys"; +export type Type50 = "GetVariableKeys"; export type Key10 = string; -export type DagId16 = string; +export type DagId17 = string; export type RunId9 = string; export type TaskId3 = string; export type MapIndex5 = number | null; export type IncludePriorDates = boolean; -export type Type50 = "GetXCom"; +export type Type51 = "GetXCom"; export type Key11 = string; -export type DagId17 = string; +export type DagId18 = string; export type RunId10 = string; export type TaskId4 = string; -export type Type51 = "GetXComCount"; +export type Type52 = "GetXComCount"; export type Key12 = string; -export type DagId18 = string; +export type DagId19 = string; export type RunId11 = string; export type TaskId5 = string; export type Offset1 = number; -export type Type52 = "GetXComSequenceItem"; +export type Type53 = "GetXComSequenceItem"; export type Key13 = string; -export type DagId19 = string; +export type DagId20 = string; export type RunId12 = string; export type TaskId6 = string; export type Start = number | null; export type Stop = number | null; export type Step = number | null; export type IncludePriorDates1 = boolean; -export type Type53 = "GetXComSequenceSlice"; +export type Type54 = "GetXComSequenceSlice"; export type TiId7 = string; /** * @minItems 1 @@ -495,21 +514,21 @@ export type Params1 = { [k: string]: unknown; } | null; export type AssignedUsers1 = HITLUser[] | null; -export type Type54 = "HITLDetailRequestResult"; +export type Type55 = "HITLDetailRequestResult"; export type InactiveAssets = AssetProfile[] | null; -export type Type55 = "InactiveAssetsResult"; +export type Type56 = "InactiveAssetsResult"; export type Name12 = string | null; -export type Type56 = "MaskSecret"; +export type Type57 = "MaskSecret"; export type Ok = boolean; -export type Type57 = "OKResponse"; +export type Type58 = "OKResponse"; export type DataIntervalStart3 = string | null; export type DataIntervalEnd3 = string | null; export type StartDate4 = string | null; export type EndDate3 = string | null; -export type Type58 = "PrevSuccessfulDagRunResult"; -export type Type59 = "PreviousDagRunResult"; +export type Type59 = "PrevSuccessfulDagRunResult"; +export type Type60 = "PreviousDagRunResult"; export type TaskId7 = string; -export type DagId20 = string; +export type DagId21 = string; export type RunId13 = string; export type LogicalDate5 = string | null; export type StartDate5 = string | null; @@ -518,51 +537,51 @@ export type State4 = string | null; export type TryNumber2 = number; export type MapIndex6 = number | null; export type Duration = number | null; -export type Type60 = "PreviousTIResult"; +export type Type61 = "PreviousTIResult"; export type Key14 = string; export type Value1 = string | null; export type Description = string | null; -export type Type61 = "PutVariable"; +export type Type62 = "PutVariable"; export type State5 = "up_for_reschedule" | null; export type RescheduleDate = string; export type EndDate5 = string; -export type Type62 = "RescheduleTask"; -export type Type63 = "ResendLoggingFD"; +export type Type63 = "RescheduleTask"; +export type Type64 = "ResendLoggingFD"; export type State6 = "up_for_retry" | null; export type EndDate6 = string; export type RenderedMapIndex2 = string | null; export type RetryDelaySeconds = number | null; export type RetryReason = string | null; -export type Type64 = "RetryTask"; -export type Type65 = "SentFDs"; +export type Type65 = "RetryTask"; +export type Type66 = "SentFDs"; export type Fds = number[]; export type Name13 = string; export type Key15 = string; -export type Type66 = "SetAssetStateStoreByName"; +export type Type67 = "SetAssetStateStoreByName"; export type Uri9 = string; export type Key16 = string; -export type Type67 = "SetAssetStateStoreByUri"; -export type Type68 = "SetRenderedFields"; +export type Type68 = "SetAssetStateStoreByUri"; +export type Type69 = "SetRenderedFields"; export type RenderedMapIndex3 = string; -export type Type69 = "SetRenderedMapIndex"; +export type Type70 = "SetRenderedMapIndex"; export type TiId8 = string; export type Key17 = string; export type ExpiresAt = string | null; -export type Type70 = "SetTaskStateStore"; +export type Type71 = "SetTaskStateStore"; export type Key18 = string; -export type DagId21 = string; +export type DagId22 = string; export type RunId14 = string; export type TaskId8 = string; export type MapIndex7 = number | null; export type DagResult1 = boolean; export type MappedLength = number | null; -export type Type71 = "SetXCom"; +export type Type72 = "SetXCom"; export type Tasks = (string | [unknown, unknown])[]; -export type Type72 = "SkipDownstreamTasks"; +export type Type73 = "SkipDownstreamTasks"; export type DagRelPath = string; export type StartDate6 = string; export type SentryIntegration = string; -export type Type73 = "StartupDetails"; +export type Type74 = "StartupDetails"; export type State7 = "success" | null; export type EndDate7 = string; export type TaskOutlets = AssetProfile[] | null; @@ -572,21 +591,21 @@ export type OutletEvents = }[] | null; export type RenderedMapIndex4 = string | null; -export type Type74 = "SucceedTask"; +export type Type75 = "SucceedTask"; export type Count1 = number; -export type Type75 = "TICount"; +export type Type76 = "TICount"; export type Breadcrumbs = { [k: string]: unknown; }[]; -export type Type76 = "TaskBreadcrumbsResult"; +export type Type77 = "TaskBreadcrumbsResult"; export type StartDate7 = string | null; -export type Type77 = "TaskRescheduleStartDate"; +export type Type78 = "TaskRescheduleStartDate"; export type State8 = "failed" | "skipped" | "removed"; export type EndDate8 = string | null; -export type Type78 = "TaskState"; +export type Type79 = "TaskState"; export type RenderedMapIndex5 = string | null; -export type Type79 = "TaskStateStoreResult"; -export type Type80 = "TaskStatesResult"; +export type Type80 = "TaskStateStoreResult"; +export type Type81 = "TaskStatesResult"; export type LogicalDate6 = string | null; export type RunAfter2 = string | null; export type Conf2 = { @@ -595,9 +614,9 @@ export type Conf2 = { export type ResetDagRun = boolean | null; export type PartitionKey7 = string | null; export type Note2 = string | null; -export type DagId22 = string; +export type DagId23 = string; export type DagRunId = string; -export type Type81 = "TriggerDagRun"; +export type Type82 = "TriggerDagRun"; export type TiId9 = string; /** * @minItems 1 @@ -606,22 +625,22 @@ export type ChosenOptions = [string, ...string[]]; export type ParamsInput = { [k: string]: unknown; } | null; -export type Type82 = "UpdateHITLDetail"; +export type Type83 = "UpdateHITLDetail"; export type TiId10 = string; -export type Type83 = "ValidateInletsAndOutlets"; +export type Type84 = "ValidateInletsAndOutlets"; export type Keys = string[]; export type TotalEntries = number; -export type Type84 = "VariableKeysResult"; +export type Type85 = "VariableKeysResult"; export type Key19 = string; export type Value2 = string | null; -export type Type85 = "VariableResult"; +export type Type86 = "VariableResult"; export type Len = number; -export type Type86 = "XComCountResponse"; +export type Type87 = "XComCountResponse"; export type Key20 = string; -export type Type87 = "XComResult"; -export type Type88 = "XComSequenceIndexResult"; +export type Type88 = "XComResult"; +export type Type89 = "XComSequenceIndexResult"; export type Root = JsonValue[]; -export type Type89 = "XComSequenceSliceResult"; +export type Type90 = "XComSequenceSliceResult"; export interface SupervisorWireSchema {} /** @@ -964,7 +983,23 @@ export interface DagFileParseRequest { bundle_path: BundlePath; bundle_name: BundleName1; callback_requests?: CallbackRequests; - type?: Type15; + type?: Type16; +} +/** + * Store skipped intervals callback data for execution by the Dag processor. + * + * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema + * via the `definition` "DagSkippedIntervalsCallbackRequest". + */ +export interface DagSkippedIntervalsCallbackRequest { + filepath: Filepath1; + bundle_name: BundleName2; + bundle_version: BundleVersion1; + version_data?: VersionData2; + msg?: Msg1; + dag_id: DagId4; + skipped_range: SkippedRange; + type?: Type13; } /** * Task callback status information. @@ -976,15 +1011,15 @@ export interface DagFileParseRequest { * via the `definition` "TaskCallbackRequest". */ export interface TaskCallbackRequest { - filepath: Filepath1; - bundle_name: BundleName2; - bundle_version: BundleVersion1; - version_data?: VersionData2; - msg?: Msg1; + filepath: Filepath2; + bundle_name: BundleName3; + bundle_version: BundleVersion2; + version_data?: VersionData3; + msg?: Msg2; ti: TaskInstance; task_callback_type?: TaskInstanceState | null; context_from_server?: TIRunContext | null; - type?: Type13; + type?: Type14; } /** * Response schema for TaskInstance run context. @@ -1037,15 +1072,15 @@ export interface ConnectionResponse { * via the `definition` "EmailRequest". */ export interface EmailRequest { - filepath: Filepath2; - bundle_name: BundleName3; - bundle_version: BundleVersion2; - version_data?: VersionData3; - msg?: Msg2; + filepath: Filepath3; + bundle_name: BundleName4; + bundle_version: BundleVersion3; + version_data?: VersionData4; + msg?: Msg3; ti: TaskInstance; email_type?: EmailType; context_from_server: TIRunContext; - type?: Type14; + type?: Type15; } /** * Result of DAG File Parsing. @@ -1061,7 +1096,7 @@ export interface DagFileParsingResult { serialized_dags: SerializedDags; warnings?: Warnings; import_errors?: ImportErrors; - type?: Type16; + type?: Type17; } /** * Lazily build information from the serialized DAG structure. @@ -1084,22 +1119,22 @@ export interface Data { * via the `definition` "DagResult". */ export interface DagResult { - dag_id: DagId4; + dag_id: DagId5; is_paused: IsPaused; - bundle_name?: BundleName4; - bundle_version?: BundleVersion3; + bundle_name?: BundleName5; + bundle_version?: BundleVersion4; relative_fileloc?: RelativeFileloc; owners?: Owners; tags: Tags; next_dagrun?: NextDagrun; - type?: Type17; + type?: Type18; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "DagRunResult". */ export interface DagRunResult { - dag_id: DagId5; + dag_id: DagId6; run_id: RunId4; logical_date?: LogicalDate2; data_interval_start?: DataIntervalStart2; @@ -1117,7 +1152,7 @@ export interface DagRunResult { partition_date?: PartitionDate1; note?: Note1; team_name?: TeamName1; - type?: Type18; + type?: Type19; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1125,7 +1160,7 @@ export interface DagRunResult { */ export interface DagRunStateResult { state: DagRunState; - type?: Type19; + type?: Type20; } /** * Update a task instance state to deferred. @@ -1142,7 +1177,7 @@ export interface DeferTask { next_method: NextMethod2; next_kwargs?: NextKwargs2; rendered_map_index?: RenderedMapIndex1; - type?: Type20; + type?: Type21; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1151,7 +1186,7 @@ export interface DeferTask { export interface DeleteAssetStateStoreByName { name: Name8; key: Key1; - type?: Type21; + type?: Type22; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1160,7 +1195,7 @@ export interface DeleteAssetStateStoreByName { export interface DeleteAssetStateStoreByUri { uri: Uri5; key: Key2; - type?: Type22; + type?: Type23; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1169,7 +1204,7 @@ export interface DeleteAssetStateStoreByUri { export interface DeleteTaskStateStore { ti_id: TiId2; key: Key3; - type?: Type23; + type?: Type24; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1177,7 +1212,7 @@ export interface DeleteTaskStateStore { */ export interface DeleteVariable { key: Key4; - type?: Type24; + type?: Type25; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1185,11 +1220,11 @@ export interface DeleteVariable { */ export interface DeleteXCom { key: Key5; - dag_id: DagId6; + dag_id: DagId7; run_id: RunId5; task_id: TaskId1; map_index?: MapIndex1; - type?: Type25; + type?: Type26; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1198,7 +1233,7 @@ export interface DeleteXCom { export interface ErrorResponse { error?: ErrorType; detail?: Detail; - type?: Type26; + type?: Type27; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1206,7 +1241,7 @@ export interface ErrorResponse { */ export interface GetAssetByName { name: Name9; - type?: Type27; + type?: Type28; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1214,7 +1249,7 @@ export interface GetAssetByName { */ export interface GetAssetByUri { uri: Uri6; - type?: Type28; + type?: Type29; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1230,7 +1265,7 @@ export interface GetAssetEventByAsset { partition_key?: PartitionKey5; partition_key_regexp_pattern?: PartitionKeyRegexpPattern; extra?: Extra7; - type?: Type29; + type?: Type30; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1245,7 +1280,7 @@ export interface GetAssetEventByAssetAlias { partition_key?: PartitionKey6; partition_key_regexp_pattern?: PartitionKeyRegexpPattern1; extra?: Extra8; - type?: Type30; + type?: Type31; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1254,7 +1289,7 @@ export interface GetAssetEventByAssetAlias { export interface GetAssetStateStoreByName { name: Name11; key: Key6; - type?: Type31; + type?: Type32; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1263,7 +1298,7 @@ export interface GetAssetStateStoreByName { export interface GetAssetStateStoreByUri { uri: Uri8; key: Key7; - type?: Type32; + type?: Type33; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1271,7 +1306,7 @@ export interface GetAssetStateStoreByUri { */ export interface GetAssetsByAlias { alias_name: AliasName1; - type?: Type33; + type?: Type34; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1279,44 +1314,44 @@ export interface GetAssetsByAlias { */ export interface GetConnection { conn_id: ConnId2; - type?: Type34; + type?: Type35; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "GetDRCount". */ export interface GetDRCount { - dag_id: DagId7; + dag_id: DagId8; logical_dates?: LogicalDates; run_ids?: RunIds; states?: States; - type?: Type35; + type?: Type36; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "GetDag". */ export interface GetDag { - dag_id: DagId8; - type?: Type36; + dag_id: DagId9; + type?: Type37; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "GetDagRun". */ export interface GetDagRun { - dag_id: DagId9; + dag_id: DagId10; run_id: RunId6; - type?: Type37; + type?: Type38; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "GetDagRunState". */ export interface GetDagRunState { - dag_id: DagId10; + dag_id: DagId11; run_id: RunId7; - type?: Type38; + type?: Type39; } /** * Get the response content part of a Human-in-the-loop response. @@ -1326,7 +1361,7 @@ export interface GetDagRunState { */ export interface GetHITLDetailResponse { ti_id: TiId3; - type?: Type39; + type?: Type40; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1334,17 +1369,17 @@ export interface GetHITLDetailResponse { */ export interface GetPrevSuccessfulDagRun { ti_id: TiId4; - type?: Type40; + type?: Type41; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "GetPreviousDagRun". */ export interface GetPreviousDagRun { - dag_id: DagId11; + dag_id: DagId12; logical_date: LogicalDate3; state?: State3; - type?: Type41; + type?: Type42; } /** * Request to get previous task instance. @@ -1353,35 +1388,35 @@ export interface GetPreviousDagRun { * via the `definition` "GetPreviousTI". */ export interface GetPreviousTI { - dag_id: DagId12; + dag_id: DagId13; task_id: TaskId2; logical_date?: LogicalDate4; map_index?: MapIndex2; state?: TaskInstanceState | null; - type?: Type42; + type?: Type43; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "GetTICount". */ export interface GetTICount { - dag_id: DagId13; + dag_id: DagId14; map_index?: MapIndex3; task_ids?: TaskIds; task_group_id?: TaskGroupId; logical_dates?: LogicalDates1; run_ids?: RunIds1; states?: States1; - type?: Type43; + type?: Type44; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "GetTaskBreadcrumbs". */ export interface GetTaskBreadcrumbs { - dag_id: DagId14; + dag_id: DagId15; run_id: RunId8; - type?: Type44; + type?: Type45; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1390,7 +1425,7 @@ export interface GetTaskBreadcrumbs { export interface GetTaskRescheduleStartDate { ti_id: TiId5; try_number?: TryNumber1; - type?: Type45; + type?: Type46; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1399,20 +1434,20 @@ export interface GetTaskRescheduleStartDate { export interface GetTaskStateStore { ti_id: TiId6; key: Key8; - type?: Type46; + type?: Type47; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "GetTaskStates". */ export interface GetTaskStates { - dag_id: DagId15; + dag_id: DagId16; map_index?: MapIndex4; task_ids?: TaskIds1; task_group_id?: TaskGroupId1; logical_dates?: LogicalDates2; run_ids?: RunIds2; - type?: Type47; + type?: Type48; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1420,7 +1455,7 @@ export interface GetTaskStates { */ export interface GetVariable { key: Key9; - type?: Type48; + type?: Type49; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1430,7 +1465,7 @@ export interface GetVariableKeys { prefix?: Prefix; limit?: Limit2; offset?: Offset; - type?: Type49; + type?: Type50; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1438,12 +1473,12 @@ export interface GetVariableKeys { */ export interface GetXCom { key: Key10; - dag_id: DagId16; + dag_id: DagId17; run_id: RunId9; task_id: TaskId3; map_index?: MapIndex5; include_prior_dates?: IncludePriorDates; - type?: Type50; + type?: Type51; } /** * Get the number of (mapped) XCom values available. @@ -1453,10 +1488,10 @@ export interface GetXCom { */ export interface GetXComCount { key: Key11; - dag_id: DagId17; + dag_id: DagId18; run_id: RunId10; task_id: TaskId4; - type?: Type51; + type?: Type52; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1464,11 +1499,11 @@ export interface GetXComCount { */ export interface GetXComSequenceItem { key: Key12; - dag_id: DagId18; + dag_id: DagId19; run_id: RunId11; task_id: TaskId5; offset: Offset1; - type?: Type52; + type?: Type53; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1476,14 +1511,14 @@ export interface GetXComSequenceItem { */ export interface GetXComSequenceSlice { key: Key13; - dag_id: DagId19; + dag_id: DagId20; run_id: RunId12; task_id: TaskId6; start: Start; stop: Stop; step: Step; include_prior_dates?: IncludePriorDates1; - type?: Type53; + type?: Type54; } /** * Response to CreateHITLDetailPayload request. @@ -1500,7 +1535,7 @@ export interface HITLDetailRequestResult { multiple?: Multiple1; params?: Params1; assigned_users?: AssignedUsers1; - type?: Type54; + type?: Type55; } /** * Response of InactiveAssets requests. @@ -1510,7 +1545,7 @@ export interface HITLDetailRequestResult { */ export interface InactiveAssetsResult { inactive_assets?: InactiveAssets; - type?: Type55; + type?: Type56; } /** * Add a new value to be redacted in task logs. @@ -1521,7 +1556,7 @@ export interface InactiveAssetsResult { export interface MaskSecret { value: JsonValue; name?: Name12; - type?: Type56; + type?: Type57; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1529,7 +1564,7 @@ export interface MaskSecret { */ export interface OKResponse { ok: Ok; - type?: Type57; + type?: Type58; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1540,7 +1575,7 @@ export interface PrevSuccessfulDagRunResult { data_interval_end?: DataIntervalEnd3; start_date?: StartDate4; end_date?: EndDate3; - type?: Type58; + type?: Type59; } /** * Response containing previous Dag run information. @@ -1550,7 +1585,7 @@ export interface PrevSuccessfulDagRunResult { */ export interface PreviousDagRunResult { dag_run?: DagRun | null; - type?: Type59; + type?: Type60; } /** * Schema for response with previous TaskInstance information. @@ -1560,7 +1595,7 @@ export interface PreviousDagRunResult { */ export interface PreviousTIResponse { task_id: TaskId7; - dag_id: DagId20; + dag_id: DagId21; run_id: RunId13; logical_date?: LogicalDate5; start_date?: StartDate5; @@ -1578,7 +1613,7 @@ export interface PreviousTIResponse { */ export interface PreviousTIResult { task_instance?: PreviousTIResponse | null; - type?: Type60; + type?: Type61; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1588,7 +1623,7 @@ export interface PutVariable { key: Key14; value: Value1; description: Description; - type?: Type61; + type?: Type62; } /** * Update a task instance state to reschedule/up_for_reschedule. @@ -1600,14 +1635,14 @@ export interface RescheduleTask { state?: State5; reschedule_date: RescheduleDate; end_date: EndDate5; - type?: Type62; + type?: Type63; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "ResendLoggingFD". */ export interface ResendLoggingFD { - type?: Type63; + type?: Type64; } /** * Update a task instance state to up_for_retry. @@ -1621,14 +1656,14 @@ export interface RetryTask { rendered_map_index?: RenderedMapIndex2; retry_delay_seconds?: RetryDelaySeconds; retry_reason?: RetryReason; - type?: Type64; + type?: Type65; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema * via the `definition` "SentFDs". */ export interface SentFDs { - type?: Type65; + type?: Type66; fds: Fds; } /** @@ -1639,7 +1674,7 @@ export interface SetAssetStateStoreByName { name: Name13; key: Key15; value: JsonValue; - type?: Type66; + type?: Type67; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1649,7 +1684,7 @@ export interface SetAssetStateStoreByUri { uri: Uri9; key: Key16; value: JsonValue; - type?: Type67; + type?: Type68; } /** * Payload for setting RTIF for a task instance. @@ -1659,7 +1694,7 @@ export interface SetAssetStateStoreByUri { */ export interface SetRenderedFields { rendered_fields: RenderedFields; - type?: Type68; + type?: Type69; } export interface RenderedFields { [k: string]: JsonValue; @@ -1672,7 +1707,7 @@ export interface RenderedFields { */ export interface SetRenderedMapIndex { rendered_map_index: RenderedMapIndex3; - type?: Type69; + type?: Type70; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1683,7 +1718,7 @@ export interface SetTaskStateStore { key: Key17; value: JsonValue; expires_at: ExpiresAt; - type?: Type70; + type?: Type71; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1692,13 +1727,13 @@ export interface SetTaskStateStore { export interface SetXCom { key: Key18; value: JsonValue; - dag_id: DagId21; + dag_id: DagId22; run_id: RunId14; task_id: TaskId8; map_index?: MapIndex7; dag_result?: DagResult1; mapped_length?: MappedLength; - type?: Type71; + type?: Type72; } /** * Update state of downstream tasks within a task instance to 'skipped', while updating current task to success state. @@ -1708,7 +1743,7 @@ export interface SetXCom { */ export interface SkipDownstreamTasks { tasks: Tasks; - type?: Type72; + type?: Type73; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1721,7 +1756,7 @@ export interface StartupDetails { start_date: StartDate6; ti_context: TIRunContext; sentry_integration: SentryIntegration; - type?: Type73; + type?: Type74; } /** * Update a task's state to success. Includes task_outlets and outlet_events for registering asset events. @@ -1735,7 +1770,7 @@ export interface SucceedTask { task_outlets?: TaskOutlets; outlet_events?: OutletEvents; rendered_map_index?: RenderedMapIndex4; - type?: Type74; + type?: Type75; } /** * Response containing count of Task Instances matching certain filters. @@ -1745,7 +1780,7 @@ export interface SucceedTask { */ export interface TICount { count: Count1; - type?: Type75; + type?: Type76; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1753,7 +1788,7 @@ export interface TICount { */ export interface TaskBreadcrumbsResult { breadcrumbs: Breadcrumbs; - type?: Type76; + type?: Type77; } /** * Response containing the first reschedule date for a task instance. @@ -1763,7 +1798,7 @@ export interface TaskBreadcrumbsResult { */ export interface TaskRescheduleStartDate { start_date: StartDate7; - type?: Type77; + type?: Type78; } /** * Update a task's state. @@ -1778,7 +1813,7 @@ export interface TaskRescheduleStartDate { export interface TaskState { state: State8; end_date?: EndDate8; - type?: Type78; + type?: Type79; rendered_map_index?: RenderedMapIndex5; } /** @@ -1789,7 +1824,7 @@ export interface TaskState { */ export interface TaskStateStoreResult { value: JsonValue; - type?: Type79; + type?: Type80; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1797,7 +1832,7 @@ export interface TaskStateStoreResult { */ export interface TaskStatesResult { task_states: TaskStates; - type?: Type80; + type?: Type81; } export interface TaskStates { [k: string]: unknown; @@ -1813,9 +1848,9 @@ export interface TriggerDagRun { reset_dag_run?: ResetDagRun; partition_key?: PartitionKey7; note?: Note2; - dag_id: DagId22; + dag_id: DagId23; run_id: DagRunId; - type?: Type81; + type?: Type82; } /** * Update the response content part of an existing Human-in-the-loop response. @@ -1827,7 +1862,7 @@ export interface UpdateHITLDetail { ti_id: TiId9; chosen_options: ChosenOptions; params_input?: ParamsInput; - type?: Type82; + type?: Type83; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1835,7 +1870,7 @@ export interface UpdateHITLDetail { */ export interface ValidateInletsAndOutlets { ti_id: TiId10; - type?: Type83; + type?: Type84; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1844,7 +1879,7 @@ export interface ValidateInletsAndOutlets { export interface VariableKeysResult { keys: Keys; total_entries: TotalEntries; - type?: Type84; + type?: Type85; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1853,7 +1888,7 @@ export interface VariableKeysResult { export interface VariableResult { key: Key19; value?: Value2; - type?: Type85; + type?: Type86; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1861,7 +1896,7 @@ export interface VariableResult { */ export interface XComCountResponse { len: Len; - type?: Type86; + type?: Type87; } /** * Response to ReadXCom request. @@ -1872,7 +1907,7 @@ export interface XComCountResponse { export interface XComResult { key: Key20; value: JsonValue; - type?: Type87; + type?: Type88; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1880,7 +1915,7 @@ export interface XComResult { */ export interface XComSequenceIndexResult { root: JsonValue; - type?: Type88; + type?: Type89; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1888,7 +1923,7 @@ export interface XComSequenceIndexResult { */ export interface XComSequenceSliceResult { root: Root; - type?: Type89; + type?: Type90; } /** Cadwyn schema version this SDK was generated against. @@ -1896,4 +1931,4 @@ export interface XComSequenceSliceResult { * (e.g. bundle metadata) and runs the migrator accordingly. * Exposed so the SDK author / operator can confirm which schema * version their build is pinned to. */ -export const SUPERVISOR_API_VERSION = "2026-06-16" as const; +export const SUPERVISOR_API_VERSION = "2026-06-23" as const;