diff --git a/airflow-core/docs/administration-and-deployment/listeners.rst b/airflow-core/docs/administration-and-deployment/listeners.rst index cf653f2fbe5e7..55c839059a16b 100644 --- a/airflow-core/docs/administration-and-deployment/listeners.rst +++ b/airflow-core/docs/administration-and-deployment/listeners.rst @@ -64,6 +64,24 @@ 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_dag_skipped_intervals`` 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. Like the callback, it does not fire on a Dag's first +automated run (no previous run to compare against). + +- ``on_dag_skipped_intervals`` + +.. exampleinclude:: /../src/airflow/example_dags/plugins/event_listener.py + :language: python + :start-after: [START howto_listen_dag_skipped_intervals] + :end-before: [END howto_listen_dag_skipped_intervals] + TaskInstance State Change Events -------------------------------- @@ -207,3 +225,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_dag_skipped_intervals`` | 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..99495148a2426 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,57 @@ 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. It does not fire on a Dag's first automated run (no previous +run to compare against), even when ``catchup=False`` leaves intervals after ``start_date`` +uncreated. + +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 9db308ddd3c72..f505747dc47e1 100644 --- a/airflow-core/docs/core-concepts/dag-run.rst +++ b/airflow-core/docs/core-concepts/dag-run.rst @@ -128,6 +128,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_dag_skipped_intervals`` 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..9278ef40a975a 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_dag_skipped_intervals] + :end-before: [END howto_listen_dag_skipped_intervals] + 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..c2f778eb17c11 --- /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_dag_skipped_intervals`` 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/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index 343d72c5e002d..eecd1d3607c72 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -48,7 +48,7 @@ from airflow._shared.observability.metrics.stats import normalize_name_for_stats from airflow._shared.timezones import timezone from airflow.api_fastapi.execution_api.in_process import InProcessExecutionAPI -from airflow.callbacks.callback_requests import DagCallbackRequest +from airflow.callbacks.callback_requests import DagCallbackRequest, DagSkippedIntervalsCallbackRequest from airflow.configuration import conf from airflow.dag_processing.bundles.base import ( BundleUsageTrackingManager, @@ -821,6 +821,8 @@ def _add_callback_to_queue(self, request: CallbackRequest) -> None: def _log_dropped_lang_sdk_callback(self, request: CallbackRequest) -> None: if isinstance(request, DagCallbackRequest): target = f"dag_id={request.dag_id} run_id={request.run_id}" + elif isinstance(request, DagSkippedIntervalsCallbackRequest): + target = f"dag_id={request.dag_id}" else: ti = request.ti target = f"dag_id={ti.dag_id} run_id={ti.run_id} task_id={ti.task_id}" diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index e6f3f422a0abf..14bcd30397f39 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -33,6 +33,7 @@ from airflow.callbacks.callback_requests import ( CallbackRequest, DagCallbackRequest, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) @@ -89,7 +90,7 @@ 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.sdk.execution_time.supervisor import RequestHandler, RequestResult, ResponseSent @@ -342,8 +343,10 @@ def _execute_callbacks( "run_id": request.ti.run_id, "ti_id": str(request.ti.id), } - else: + elif isinstance(request, DagCallbackRequest): log_extra = {"dag_id": request.dag_id, "run_id": request.run_id} + else: + log_extra = {"dag_id": request.dag_id} # context_from_server can carry user-supplied run conf, and the masker cannot # redact inside an already-serialized string, so keep it out of log payloads. request_json = request.to_json(exclude={"context_from_server"}) @@ -360,6 +363,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) except Exception: @@ -432,6 +437,44 @@ 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=prune_dict( + { + "dag_id": request.dag_id, + "team_name": ( + DagModel.get_team_name(request.dag_id) + if conf.getboolean("core", "multi_team") + else None + ), + } + ), + ) + + 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..51ef38375eb5b 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_dag_skipped_intervals] +@hookimpl +def on_dag_skipped_intervals(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_dag_skipped_intervals] diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index 40e1fa976ffeb..dfed8a993922b 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 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 ( @@ -66,6 +66,7 @@ from airflow.assets.evaluation import AssetEvaluator from airflow.callbacks.callback_requests import ( DagCallbackRequest, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) @@ -77,6 +78,7 @@ from airflow.executors.workloads.types import TaskInstanceUuid 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, @@ -123,7 +125,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.triggers.base import TriggerEvent from airflow.utils.event_scheduler import EventScheduler from airflow.utils.helpers import prune_dict @@ -163,6 +170,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:""" @@ -2177,9 +2192,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() @@ -2212,6 +2228,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() @@ -2639,7 +2660,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) @@ -2653,7 +2676,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], @@ -2664,6 +2687,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.""" @@ -2695,8 +2720,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_dag_skipped_intervals.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 @@ -2804,6 +2879,12 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) - logical_date = next_info.logical_date partition_key = next_info.partition_key run_after = next_info.run_after + skipped_summary = self._collect_skipped_intervals( + serdag=serdag, + new_info=next_info, + session=session, + listener_has_impls=listener_has_impls, + ) created_run = serdag.create_dagrun( run_id=serdag.timetable.generate_run_id( run_type=DagRunType.SCHEDULED, @@ -2835,6 +2916,20 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) - records=consumed_asset_records, dag_id=dag_model.dag_id, session=session ) + 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 @@ -2849,6 +2944,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_dag_skipped_intervals( # 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 _collect_gated_asset_events( self, *, dag: SerializedDAG, session: Session ) -> tuple[Sequence[AssetDagRunQueue], list[AssetEvent]] | None: diff --git a/airflow-core/src/airflow/listeners/listener.py b/airflow-core/src/airflow/listeners/listener.py index 06be7a1b9b908..166768e03b717 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_dag_skipped_intervals - 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..194f22ded8c07 --- /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_dag_skipped_intervals(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 c529bf95adea9..5bdefc65c0ff3 100644 --- a/airflow-core/src/airflow/serialization/definitions/dag.py +++ b/airflow-core/src/airflow/serialization/definitions/dag.py @@ -53,7 +53,7 @@ from airflow.serialization.decoders import decode_deadline_alert, resolve_deadline_alert_interval from airflow.serialization.definitions.deadline import DeadlineAlertFields, SerializedReferenceModels from airflow.serialization.definitions.param import SerializedParamsDict -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.sqlalchemy import PROHIBIT_COMMIT_ERROR_MESSAGE @@ -127,6 +127,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 @@ -541,6 +542,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 ed2251ce3dac7..51b8414ab5818 100644 --- a/airflow-core/src/airflow/serialization/schema.json +++ b/airflow-core/src/airflow/serialization/schema.json @@ -237,6 +237,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 f5fd8c118b9b3..f1aaed78db653 100644 --- a/airflow-core/src/airflow/serialization/serialized_objects.py +++ b/airflow-core/src/airflow/serialization/serialized_objects.py @@ -140,7 +140,9 @@ _CALLBACK_TYPES = ("execute", "failure", "success", "retry", "skipped") _OPERATOR_CALLBACK_FIELDS = frozenset(f"on_{x}_callback" for x in _CALLBACK_TYPES) _HAS_CALLBACK_FIELDS = frozenset(f"has_on_{x}_callback" for x in _CALLBACK_TYPES) -_DAG_CALLBACK_FIELDS = frozenset({"has_on_success_callback", "has_on_failure_callback"}) +_DAG_CALLBACK_FIELDS = frozenset( + {"has_on_success_callback", "has_on_failure_callback", "has_on_skipped_intervals_callback"} +) _OPERATOR_TIMEDELTA_FIELDS = frozenset({"retry_delay", "execution_timeout", "max_retry_delay"}) # Fields whose value must never be serialized: the object has no serializer, so it would # fall back to str(obj) and leak a non-deterministic memory address (a new DagVersion every @@ -2376,6 +2378,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 a2325395a08e4..6fb42c24e37c4 100644 --- a/airflow-core/src/airflow/timetables/base.py +++ b/airflow-core/src/airflow/timetables/base.py @@ -74,6 +74,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 6a82a22fce2d1..1f9bc113ec9be 100644 --- a/airflow-core/tests/unit/callbacks/test_callback_requests.py +++ b/airflow-core/tests/unit/callbacks/test_callback_requests.py @@ -32,12 +32,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 @@ -409,6 +411,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_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index 2a92835edba39..c6cd17876337f 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -47,7 +47,11 @@ from airflow._shared.timezones import timezone from airflow.api_fastapi.execution_api.datamodels.taskinstance import TaskInstance as TIDataModel -from airflow.callbacks.callback_requests import DagCallbackRequest, TaskCallbackRequest +from airflow.callbacks.callback_requests import ( + DagCallbackRequest, + DagSkippedIntervalsCallbackRequest, + TaskCallbackRequest, +) from airflow.dag_processing.bundles.base import BaseDagBundle, BundleVersion from airflow.dag_processing.bundles.manager import DagBundlesManager from airflow.dag_processing.collection import update_dag_parsing_results_in_db @@ -76,6 +80,7 @@ from airflow.sdk import DAG as SdkDAG, BaseOperator from airflow.sdk.importers import DagDefinition, DagImporterRegistry, DagImportError, DagSourceCode from airflow.serialization.serialized_objects import DagSerialization, LazyDeserializedDAG +from airflow.timetables.base import DataInterval, SkippedIntervalsSummary from airflow.utils.net import get_hostname from airflow.utils.session import create_session @@ -1746,6 +1751,20 @@ def test_create_process_keeps_the_python_parse_for_other_files(self, mock_get_lo "task_id=extract): Lang-SDK runtimes do not run callbacks", id="task", ), + pytest.param( + DagSkippedIntervalsCallbackRequest.from_summary( + filepath="dags.native", + dag_id="native_dag", + bundle_name="testing", + bundle_version="v1", + summary=SkippedIntervalsSummary( + skipped_range=DataInterval(start=DEFAULT_DATE, end=DEFAULT_DATE), + ), + ), + "Dropping DagSkippedIntervalsCallbackRequest for dags.native (dag_id=native_dag): " + "Lang-SDK runtimes do not run callbacks", + id="skipped_intervals", + ), ], ) @mock.patch.object(DagFileProcessorManager, "prepare_callback_bundle", autospec=True) diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py b/airflow-core/tests/unit/dag_processing/test_processor.py index 629cb6b1c49b1..e0b0197fd3899 100644 --- a/airflow-core/tests/unit/dag_processing/test_processor.py +++ b/airflow-core/tests/unit/dag_processing/test_processor.py @@ -48,6 +48,7 @@ CallbackRequest, DagCallbackRequest, DagRunContext, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) @@ -62,6 +63,7 @@ ToManager, _execute_callbacks, _execute_dag_callbacks, + _execute_dag_skipped_intervals_callback, _execute_email_callbacks, _execute_task_callbacks, _parse_file, @@ -91,6 +93,7 @@ from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance from airflow.sdk.importers import DagSourceCode from airflow.serialization.serialized_objects import DagSerialization +from airflow.timetables.base import DataInterval, SkippedIntervalsSummary from airflow.utils.session import create_session from airflow.utils.state import TaskInstanceState @@ -99,6 +102,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 @@ -980,6 +1007,22 @@ def make_request(dag_id): assert log.exception.call_args.kwargs["dag_id"] == "removed_dag" assert "context_from_server" not in log.exception.call_args.kwargs["request"] + 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""" @@ -1556,6 +1599,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 7118d08edc61f..7793e16462142 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -53,6 +53,7 @@ from airflow.callbacks.callback_requests import ( DagCallbackRequest, DagRunContext, + DagSkippedIntervalsCallbackRequest, EmailRequest, TaskCallbackRequest, ) @@ -12060,6 +12061,366 @@ 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 == [] + + @conf_vars({("scheduler", "create_cron_data_intervals"): "False"}) + def test_create_dag_runs_emits_skipped_intervals_for_multi_day_cron_gap(self, session, dag_maker): + """CronTriggerTimetable gaps must fire skip callbacks via _create_dag_runs().""" + with dag_maker( + dag_id="test_multi_day_cron_gap_skip", + schedule="@daily", + start_date=DEFAULT_DATE, + catchup=False, + on_skipped_intervals_callback=lambda ctx: None, + session=session, + ) as dag: + EmptyOperator(task_id="dummy") + + dag_maker.create_dagrun( + run_type=DagRunType.SCHEDULED, + logical_date=DEFAULT_DATE, + data_interval=DataInterval.exact(DEFAULT_DATE), + state=State.SUCCESS, + ) + + new_start = DEFAULT_DATE + timedelta(days=3) + dag_model = dag_maker.dag_model + dag_model.next_dagrun = new_start + dag_model.next_dagrun_data_interval = DataInterval.exact(new_start) + dag_model.next_dagrun_create_after = new_start + 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) + 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 == (DEFAULT_DATE, new_start) + + 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..521ef59fe5745 --- /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_dag_skipped_intervals(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 d40f36ebf6d33..d3f141ce5776e 100644 --- a/airflow-core/tests/unit/serialization/test_dag_serialization.py +++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py @@ -1585,6 +1585,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", } @@ -2537,6 +2538,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/internal/genspec/authoring.go b/go-sdk/internal/genspec/authoring.go index 463eef2e44021..025392e7ce244 100644 --- a/go-sdk/internal/genspec/authoring.go +++ b/go-sdk/internal/genspec/authoring.go @@ -72,27 +72,28 @@ var authoringShapes = map[string]authoringShape{ var dagShape = authoringShape{ doc: "DagSpec holds the attributes of a Dag other than its dag_id. Dag takes one.", exclude: map[string]string{ - "dag_id": "a positional parameter of airflow.Dag, not a spec field", - "fileloc": "the path of the Dag file, which the bundle fills in", - "relative_fileloc": "the path of the Dag file, which the bundle fills in", - "_processor_dags_folder": "the Dag processor's own folder, filled in at parse time", - "bundle_name": "the name of the bundle that carries the Dag, not the Dag's", - "tasks": "the tasks dag.Task registers", - "task_group": "the groups dag.TaskGroup registers", - "edge_info": "the labels airflow.Label carries into an edge verb", - "dag_dependencies": "derived from the edges and the assets a Dag declares", - "timezone": "always UTC for a Go Dag, even when StartDate has another location", - "timetable": "the serialized form of Schedule, which is injected instead", - "allowed_run_types": "no Go authoring type yet: a list of DagRunType values", - "_concurrency": "the pre-2.2 spelling of MaxActiveTasks", - "has_on_success_callback": "derived from whether a callback is registered", - "has_on_failure_callback": "derived from whether a callback is registered", - "params": "no Go authoring type yet: a param carries a schema of its own", - "default_args": "no Go authoring type yet: the values are arbitrary and untyped", - "access_control": "no Go authoring type yet, and it is deprecated in Airflow 3", - "owner_links": "no Go authoring type yet: an object of arbitrary link targets", - "deadline": "no Go authoring type yet: a serialized deadline reference", - "rerun_with_latest_version": "no Go authoring type yet: the tri-state a null allows", + "dag_id": "a positional parameter of airflow.Dag, not a spec field", + "fileloc": "the path of the Dag file, which the bundle fills in", + "relative_fileloc": "the path of the Dag file, which the bundle fills in", + "_processor_dags_folder": "the Dag processor's own folder, filled in at parse time", + "bundle_name": "the name of the bundle that carries the Dag, not the Dag's", + "tasks": "the tasks dag.Task registers", + "task_group": "the groups dag.TaskGroup registers", + "edge_info": "the labels airflow.Label carries into an edge verb", + "dag_dependencies": "derived from the edges and the assets a Dag declares", + "timezone": "always UTC for a Go Dag, even when StartDate has another location", + "timetable": "the serialized form of Schedule, which is injected instead", + "allowed_run_types": "no Go authoring type yet: a list of DagRunType values", + "_concurrency": "the pre-2.2 spelling of MaxActiveTasks", + "has_on_success_callback": "derived from whether a callback is registered", + "has_on_failure_callback": "derived from whether a callback is registered", + "has_on_skipped_intervals_callback": "derived from whether a callback is registered", + "params": "no Go authoring type yet: a param carries a schema of its own", + "default_args": "no Go authoring type yet: the values are arbitrary and untyped", + "access_control": "no Go authoring type yet, and it is deprecated in Airflow 3", + "owner_links": "no Go authoring type yet: an object of arbitrary link targets", + "deadline": "no Go authoring type yet: a serialized deadline reference", + "rerun_with_latest_version": "no Go authoring type yet: the tri-state a null allows", }, override: map[string]propertyOverride{ "start_date": { diff --git a/go-sdk/pkg/execution/genmodels/discriminators.gen.go b/go-sdk/pkg/execution/genmodels/discriminators.gen.go index 8736a6e2c9730..ca3455b713607 100644 --- a/go-sdk/pkg/execution/genmodels/discriminators.gen.go +++ b/go-sdk/pkg/execution/genmodels/discriminators.gen.go @@ -24,97 +24,98 @@ 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" - TypeSetExecutionTimeout = "SetExecutionTimeout" - 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" - TypeUpdateDagRunNote = "UpdateDagRunNote" - 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" + TypeSetExecutionTimeout = "SetExecutionTimeout" + 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" + TypeUpdateDagRunNote = "UpdateDagRunNote" + 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 @@ -174,6 +175,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 6e53729efe936..5d0e3f5a55e9f 100644 --- a/go-sdk/pkg/execution/genmodels/models.gen.go +++ b/go-sdk/pkg/execution/genmodels/models.gen.go @@ -640,6 +640,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"` +} + // Raw source code and its language identifier for a DAG definition. type DagSourceCode struct { // Language corresponds to the JSON schema field "language". @@ -1720,19 +1747,51 @@ type TIRunContext struct { XcomKeysToClear []string `msgpack:"xcom_keys_to_clear,omitempty"` } -type TriggerKwargs map[string]interface{} +type TaskArgBinding interface{} -type Warnings []interface{} +type TaskBreadcrumbsResult struct { + // Breadcrumbs corresponds to the JSON schema field "breadcrumbs". + Breadcrumbs []TaskBreadcrumbsResultBreadcrumbsElem `msgpack:"breadcrumbs"` -type TaskArgBinding interface{} + // Type corresponds to the JSON schema field "type". + Type string `msgpack:"type,omitempty"` +} -// 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 TaskBreadcrumbsResultBreadcrumbsElem map[string]interface{} - // Value corresponds to the JSON schema field "value". - Value interface{} `msgpack:"value"` +// 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"` } type TaskIds []string @@ -1776,69 +1835,24 @@ 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 TaskInstanceStateRemoved TaskInstanceState = "removed" - -type TaskInstanceState string +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 -type TaskBreadcrumbsResultBreadcrumbsElem map[string]interface{} - -type TaskBreadcrumbsResult struct { - // Breadcrumbs corresponds to the JSON schema field "breadcrumbs". - Breadcrumbs []TaskBreadcrumbsResultBreadcrumbsElem `msgpack:"breadcrumbs"` - - // Type corresponds to the JSON schema field "type". - Type string `msgpack:"type,omitempty"` -} - -// 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". @@ -1848,12 +1862,6 @@ 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 @@ -1877,6 +1885,12 @@ 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 { @@ -1926,6 +1940,21 @@ type TriggerDagRun struct { Type string `msgpack:"type,omitempty"` } +type VersionData map[string]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 Warnings []interface{} + +type TriggerKwargs map[string]interface{} + type UpdateDagRunNote struct { // Note corresponds to the JSON schema field "note". Note interface{} `msgpack:"note"` @@ -1937,8 +1966,6 @@ type UpdateDagRunNote struct { Type string `msgpack:"type,omitempty"` } -type VersionData map[string]interface{} - // Update the response content part of an existing Human-in-the-loop response. type UpdateHITLDetail struct { // ChosenOptions corresponds to the JSON schema field "chosen_options". diff --git a/go-sdk/schema/dag-schema.json b/go-sdk/schema/dag-schema.json index ed2251ce3dac7..51b8414ab5818 100644 --- a/go-sdk/schema/dag-schema.json +++ b/go-sdk/schema/dag-schema.json @@ -237,6 +237,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/go-sdk/schema/supervisor-schema.json b/go-sdk/schema/supervisor-schema.json index 77888c5adfdea..228624fa92bab 100644 --- a/go-sdk/schema/supervisor-schema.json +++ b/go-sdk/schema/supervisor-schema.json @@ -938,6 +938,7 @@ "discriminator": { "mapping": { "DagCallbackRequest": "#/$defs/DagCallbackRequest", + "DagSkippedIntervalsCallbackRequest": "#/$defs/DagSkippedIntervalsCallbackRequest", "EmailRequest": "#/$defs/EmailRequest", "TaskCallbackRequest": "#/$defs/TaskCallbackRequest" }, @@ -947,6 +948,9 @@ { "$ref": "#/$defs/DagCallbackRequest" }, + { + "$ref": "#/$defs/DagSkippedIntervalsCallbackRequest" + }, { "$ref": "#/$defs/TaskCallbackRequest" }, @@ -1507,6 +1511,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" + }, "DagSourceCode": { "description": "Raw source code and its language identifier for a DAG definition.", "properties": { diff --git a/java-sdk/sdk/schema/dag-schema.json b/java-sdk/sdk/schema/dag-schema.json index ed2251ce3dac7..51b8414ab5818 100644 --- a/java-sdk/sdk/schema/dag-schema.json +++ b/java-sdk/sdk/schema/dag-schema.json @@ -237,6 +237,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/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 1de617447fded..915ab84742468 100644 --- a/task-sdk/docs/api.rst +++ b/task-sdk/docs/api.rst @@ -360,6 +360,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 9e15dae85e550..30da3a9c5118e 100644 --- a/task-sdk/docs/index.rst +++ b/task-sdk/docs/index.rst @@ -99,6 +99,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 c163df525777c..605c973abac47 100644 --- a/task-sdk/src/airflow/sdk/__init__.py +++ b/task-sdk/src/airflow/sdk/__init__.py @@ -85,6 +85,7 @@ "RollupMapper", "SegmentWindow", "SkipMixin", + "SkippedIntervalsCallbackContext", "SyncCallback", "StartOfDayMapper", "StartOfHourMapper", @@ -154,7 +155,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 ( BaseDeadlineReference, @@ -301,6 +307,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 6f85470cc6819..8de306942819f 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, NotRequired, TypedDict, cast +from typing import TYPE_CHECKING, Any, Literal, NamedTuple, NotRequired, TypedDict, cast if TYPE_CHECKING: import jinja2 @@ -40,6 +40,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()) - {"loop"} diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py b/task-sdk/src/airflow/sdk/definitions/dag.py index 0594ac8065885..5168f3704ccb9 100644 --- a/task-sdk/src/airflow/sdk/definitions/dag.py +++ b/task-sdk/src/airflow/sdk/definitions/dag.py @@ -47,7 +47,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 @@ -109,6 +109,7 @@ ) DagStateChangeCallback = Callable[[Context], None] +SkippedIntervalsCallback = Callable[[SkippedIntervalsCallbackContext], None] ScheduleInterval = None | str | timedelta | relativedelta ScheduleArg = Union[ScheduleInterval, BaseTimetable, "CoreTimetable", BaseAsset, Collection[BaseAsset]] @@ -402,6 +403,13 @@ 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). Does not fire on the first automated + run (no previous run to compare against). 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., @@ -534,6 +542,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 @@ -575,6 +584,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") ) @@ -725,6 +735,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: @@ -1635,11 +1649,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", } @@ -1667,6 +1683,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 77888c5adfdea..228624fa92bab 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json +++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json @@ -938,6 +938,7 @@ "discriminator": { "mapping": { "DagCallbackRequest": "#/$defs/DagCallbackRequest", + "DagSkippedIntervalsCallbackRequest": "#/$defs/DagSkippedIntervalsCallbackRequest", "EmailRequest": "#/$defs/EmailRequest", "TaskCallbackRequest": "#/$defs/TaskCallbackRequest" }, @@ -947,6 +948,9 @@ { "$ref": "#/$defs/DagCallbackRequest" }, + { + "$ref": "#/$defs/DagSkippedIntervalsCallbackRequest" + }, { "$ref": "#/$defs/TaskCallbackRequest" }, @@ -1507,6 +1511,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" + }, "DagSourceCode": { "description": "Raw source code and its language identifier for a DAG definition.", "properties": { 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 0f35e9f0e6f2b..b0de197b44f26 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 @@ -40,6 +40,7 @@ def get_bundle() -> VersionBundle: from airflow.sdk.execution_time.schema.versions.v2026_10_30 import ( AddArgBindingsToSupervisorTIRunContext, AddDagDefinitionsToDagFileParsingResult, + AddDagSkippedIntervalsCallbackRequest, AddFirstTaskRescheduleStartDateToSupervisorTIRunContext, AddLoopRuntime, AddRegionCoordinatesToSupervisorTaskInstance, @@ -56,6 +57,7 @@ def get_bundle() -> VersionBundle: AddDagDefinitionsToDagFileParsingResult, AddRegionCoordinatesToSupervisorTaskInstance, AddFirstTaskRescheduleStartDateToSupervisorTIRunContext, + AddDagSkippedIntervalsCallbackRequest, ), Version("2026-06-16"), ) diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py index d1efcf0b8c380..910faec1bde49 100644 --- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py +++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py @@ -26,9 +26,11 @@ from __future__ import annotations -from cadwyn import VersionChange, schema +import logging -from airflow.dag_processing.processor import DagFileParsingResult # noqa: SDK002 +from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for, schema + +from airflow.dag_processing.processor import DagFileParseRequest, DagFileParsingResult # noqa: SDK002 from airflow.sdk.api.datamodels._generated import PreviousTIResponse, TaskInstance, TIRunContext from airflow.sdk.execution_time.comms import ( GetXCom, @@ -38,6 +40,10 @@ TaskState, ) +logger = logging.getLogger(__name__) + +_SKIPPED_INTERVALS_CALLBACK_TYPE = "DagSkippedIntervalsCallbackRequest" + class AddLoopRuntime(VersionChange): """Carry loop context and previous-iteration reads.""" @@ -111,3 +117,31 @@ class AddFirstTaskRescheduleStartDateToSupervisorTIRunContext(VersionChange): instructions_to_migrate_to_previous_version = ( schema(TIRunContext).field("first_task_reschedule_start_date").didnt_exist, ) + + +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/schema/dag-schema.json b/ts-sdk/schema/dag-schema.json index ed2251ce3dac7..51b8414ab5818 100644 --- a/ts-sdk/schema/dag-schema.json +++ b/ts-sdk/schema/dag-schema.json @@ -237,6 +237,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/ts-sdk/schema/supervisor-schema.json b/ts-sdk/schema/supervisor-schema.json index 77888c5adfdea..228624fa92bab 100644 --- a/ts-sdk/schema/supervisor-schema.json +++ b/ts-sdk/schema/supervisor-schema.json @@ -938,6 +938,7 @@ "discriminator": { "mapping": { "DagCallbackRequest": "#/$defs/DagCallbackRequest", + "DagSkippedIntervalsCallbackRequest": "#/$defs/DagSkippedIntervalsCallbackRequest", "EmailRequest": "#/$defs/EmailRequest", "TaskCallbackRequest": "#/$defs/TaskCallbackRequest" }, @@ -947,6 +948,9 @@ { "$ref": "#/$defs/DagCallbackRequest" }, + { + "$ref": "#/$defs/DagSkippedIntervalsCallbackRequest" + }, { "$ref": "#/$defs/TaskCallbackRequest" }, @@ -1507,6 +1511,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" + }, "DagSourceCode": { "description": "Raw source code and its language identifier for a DAG definition.", "properties": { diff --git a/ts-sdk/src/generated/supervisor.ts b/ts-sdk/src/generated/supervisor.ts index 7faf8c4877029..d55d81caee217 100644 --- a/ts-sdk/src/generated/supervisor.ts +++ b/ts-sdk/src/generated/supervisor.ts @@ -209,6 +209,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. * @@ -266,18 +280,20 @@ export type Name9 = string; export type Kind1 = "literal"; export type FromDefault = boolean; export type MultiTeam = boolean; -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[]; @@ -288,17 +304,17 @@ export type ImportErrors = { export type ParsedDefinitions = string[]; export type SourceCode = string; export type Language = string; -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; @@ -316,8 +332,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"; export type Classpath = string; export type TriggerKwargs = @@ -333,24 +349,24 @@ export type NextKwargs2 = { [k: string]: JsonValue | null; } | null; export type RenderedMapIndex1 = string | null; -export type Type20 = "DeferTask"; +export type Type21 = "DeferTask"; export type Name10 = 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 TaskId2 = string; export type MapIndex1 = number | null; -export type Type25 = "DeleteXCom"; +export type Type26 = "DeleteXCom"; /** * Error types used in the API client. */ @@ -368,7 +384,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. * @@ -387,9 +403,9 @@ export type ErrorType1 = | "GENERIC_ERROR" | "API_SERVER_ERROR"; export type Name11 = string; -export type Type27 = "GetAssetByName"; +export type Type28 = "GetAssetByName"; export type Uri6 = string; -export type Type28 = "GetAssetByUri"; +export type Type29 = "GetAssetByUri"; export type Name12 = string | null; export type Uri7 = string | null; export type After = string | null; @@ -401,7 +417,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; @@ -412,104 +428,104 @@ export type PartitionKeyRegexpPattern1 = string | null; export type Extra8 = { [k: string]: string; } | null; -export type Type30 = "GetAssetEventByAssetAlias"; +export type Type31 = "GetAssetEventByAssetAlias"; export type Name13 = 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 TaskId3 = 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 PreviousIteration = boolean; export type Key10 = string; -export type DagId16 = string; +export type DagId17 = string; export type RunId9 = string; export type TaskId4 = string; export type MapIndex5 = number | null; export type IncludePriorDates = boolean; -export type Type50 = "GetXCom"; +export type Type51 = "GetXCom"; export type PreviousIteration1 = boolean; export type Key11 = string; -export type DagId17 = string; +export type DagId18 = string; export type RunId10 = string; export type TaskId5 = string; -export type Type51 = "GetXComCount"; +export type Type52 = "GetXComCount"; export type PreviousIteration2 = boolean; export type Key12 = string; -export type DagId18 = string; +export type DagId19 = string; export type RunId11 = string; export type TaskId6 = string; export type Offset1 = number; -export type Type52 = "GetXComSequenceItem"; +export type Type53 = "GetXComSequenceItem"; export type PreviousIteration3 = boolean; export type Key13 = string; -export type DagId19 = string; +export type DagId20 = string; export type RunId12 = string; export type TaskId7 = 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 @@ -523,21 +539,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 Name14 = 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 TaskId8 = string; -export type DagId20 = string; +export type DagId21 = string; export type RunId13 = string; export type LogicalDate5 = string | null; export type StartDate5 = string | null; @@ -548,53 +564,53 @@ export type MapIndex6 = number | null; export type RegionId1 = string | null; export type RegionIndex1 = 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"; 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"; 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 Name15 = 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 = "SetAssetStateStoreByUri"; export type TimeoutSeconds = number; -export type Type68 = "SetExecutionTimeout"; -export type Type69 = "SetRenderedFields"; +export type Type69 = "SetExecutionTimeout"; +export type Type70 = "SetRenderedFields"; export type RenderedMapIndex3 = string; -export type Type70 = "SetRenderedMapIndex"; +export type Type71 = "SetRenderedMapIndex"; export type TiId8 = string; export type Key17 = string; export type ExpiresAt = string | null; -export type Type71 = "SetTaskStateStore"; +export type Type72 = "SetTaskStateStore"; export type Key18 = string; -export type DagId21 = string; +export type DagId22 = string; export type RunId14 = string; export type TaskId9 = string; export type MapIndex7 = number | null; export type DagResult1 = boolean; export type MappedLength = number | null; -export type Type72 = "SetXCom"; +export type Type73 = "SetXCom"; export type Tasks = (string | [unknown, unknown])[]; -export type Type73 = "SkipDownstreamTasks"; +export type Type74 = "SkipDownstreamTasks"; export type DagRelPath = string; export type StartDate6 = string; export type SentryIntegration = string; -export type Type74 = "StartupDetails"; +export type Type75 = "StartupDetails"; export type State7 = "success"; export type EndDate7 = string; export type TaskOutlets = AssetProfile[] | null; @@ -604,22 +620,22 @@ export type OutletEvents = }[] | null; export type RenderedMapIndex4 = string | null; -export type Type75 = "SucceedTask"; +export type Type76 = "SucceedTask"; export type Count1 = number; -export type Type76 = "TICount"; +export type Type77 = "TICount"; export type Breadcrumbs = { [k: string]: unknown; }[]; -export type Type77 = "TaskBreadcrumbsResult"; +export type Type78 = "TaskBreadcrumbsResult"; export type StartDate7 = string | null; -export type Type78 = "TaskRescheduleStartDate"; +export type Type79 = "TaskRescheduleStartDate"; export type State8 = "failed" | "skipped" | "removed"; export type EndDate8 = string | null; -export type Type79 = "TaskState"; +export type Type80 = "TaskState"; export type RenderedMapIndex5 = string | null; export type RetryReason1 = string | null; -export type Type80 = "TaskStateStoreResult"; -export type Type81 = "TaskStatesResult"; +export type Type81 = "TaskStateStoreResult"; +export type Type82 = "TaskStatesResult"; export type LogicalDate6 = string | null; export type RunAfter2 = string | null; export type Conf2 = { @@ -628,12 +644,12 @@ 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 Type82 = "TriggerDagRun"; +export type Type83 = "TriggerDagRun"; export type TiId9 = string; export type Note3 = string | null; -export type Type83 = "UpdateDagRunNote"; +export type Type84 = "UpdateDagRunNote"; export type TiId10 = string; /** * @minItems 1 @@ -642,22 +658,22 @@ export type ChosenOptions = [string, ...string[]]; export type ParamsInput = { [k: string]: unknown; } | null; -export type Type84 = "UpdateHITLDetail"; +export type Type85 = "UpdateHITLDetail"; export type TiId11 = string; -export type Type85 = "ValidateInletsAndOutlets"; +export type Type86 = "ValidateInletsAndOutlets"; export type Keys = string[]; export type TotalEntries = number; -export type Type86 = "VariableKeysResult"; +export type Type87 = "VariableKeysResult"; export type Key19 = string; export type Value2 = string | null; -export type Type87 = "VariableResult"; +export type Type88 = "VariableResult"; export type Len = number; -export type Type88 = "XComCountResponse"; +export type Type89 = "XComCountResponse"; export type Key20 = string; -export type Type89 = "XComResult"; -export type Type90 = "XComSequenceIndexResult"; +export type Type90 = "XComResult"; +export type Type91 = "XComSequenceIndexResult"; export type Root = JsonValue[]; -export type Type91 = "XComSequenceSliceResult"; +export type Type92 = "XComSequenceSliceResult"; export interface SupervisorWireSchema {} /** @@ -1018,7 +1034,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. @@ -1030,15 +1062,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. @@ -1124,15 +1156,15 @@ export interface LiteralArgBinding { * 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. @@ -1150,7 +1182,7 @@ export interface DagFileParsingResult { import_errors?: ImportErrors; parsed_definitions?: ParsedDefinitions; dag_source_codes?: DagSourceCodes; - type?: Type16; + type?: Type17; } /** * Lazily build information from the serialized DAG structure. @@ -1186,22 +1218,22 @@ export interface DagSourceCode { * 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; @@ -1219,7 +1251,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 @@ -1227,7 +1259,7 @@ export interface DagRunResult { */ export interface DagRunStateResult { state: DagRunState; - type?: Type19; + type?: Type20; } /** * Update a task instance state to deferred. @@ -1244,7 +1276,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 @@ -1253,7 +1285,7 @@ export interface DeferTask { export interface DeleteAssetStateStoreByName { name: Name10; key: Key1; - type?: Type21; + type?: Type22; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1262,7 +1294,7 @@ export interface DeleteAssetStateStoreByName { export interface DeleteAssetStateStoreByUri { uri: Uri5; key: Key2; - type?: Type22; + type?: Type23; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1271,7 +1303,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 @@ -1279,7 +1311,7 @@ export interface DeleteTaskStateStore { */ export interface DeleteVariable { key: Key4; - type?: Type24; + type?: Type25; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1287,11 +1319,11 @@ export interface DeleteVariable { */ export interface DeleteXCom { key: Key5; - dag_id: DagId6; + dag_id: DagId7; run_id: RunId5; task_id: TaskId2; map_index?: MapIndex1; - type?: Type25; + type?: Type26; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1300,7 +1332,7 @@ export interface DeleteXCom { export interface ErrorResponse { error?: ErrorType; detail?: Detail; - type?: Type26; + type?: Type27; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1308,7 +1340,7 @@ export interface ErrorResponse { */ export interface GetAssetByName { name: Name11; - type?: Type27; + type?: Type28; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1316,7 +1348,7 @@ export interface GetAssetByName { */ export interface GetAssetByUri { uri: Uri6; - type?: Type28; + type?: Type29; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1332,7 +1364,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 @@ -1347,7 +1379,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 @@ -1356,7 +1388,7 @@ export interface GetAssetEventByAssetAlias { export interface GetAssetStateStoreByName { name: Name13; key: Key6; - type?: Type31; + type?: Type32; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1365,7 +1397,7 @@ export interface GetAssetStateStoreByName { export interface GetAssetStateStoreByUri { uri: Uri8; key: Key7; - type?: Type32; + type?: Type33; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1373,7 +1405,7 @@ export interface GetAssetStateStoreByUri { */ export interface GetAssetsByAlias { alias_name: AliasName1; - type?: Type33; + type?: Type34; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1381,44 +1413,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. @@ -1428,7 +1460,7 @@ export interface GetDagRunState { */ export interface GetHITLDetailResponse { ti_id: TiId3; - type?: Type39; + type?: Type40; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1436,17 +1468,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. @@ -1455,35 +1487,35 @@ export interface GetPreviousDagRun { * via the `definition` "GetPreviousTI". */ export interface GetPreviousTI { - dag_id: DagId12; + dag_id: DagId13; task_id: TaskId3; 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 @@ -1492,7 +1524,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 @@ -1501,20 +1533,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 @@ -1522,7 +1554,7 @@ export interface GetTaskStates { */ export interface GetVariable { key: Key9; - type?: Type48; + type?: Type49; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1532,7 +1564,7 @@ export interface GetVariableKeys { prefix?: Prefix; limit?: Limit2; offset?: Offset; - type?: Type49; + type?: Type50; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1541,12 +1573,12 @@ export interface GetVariableKeys { export interface GetXCom { previous_iteration?: PreviousIteration; key: Key10; - dag_id: DagId16; + dag_id: DagId17; run_id: RunId9; task_id: TaskId4; map_index?: MapIndex5; include_prior_dates?: IncludePriorDates; - type?: Type50; + type?: Type51; } /** * Get the number of (mapped) XCom values available. @@ -1557,10 +1589,10 @@ export interface GetXCom { export interface GetXComCount { previous_iteration?: PreviousIteration1; key: Key11; - dag_id: DagId17; + dag_id: DagId18; run_id: RunId10; task_id: TaskId5; - type?: Type51; + type?: Type52; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1569,11 +1601,11 @@ export interface GetXComCount { export interface GetXComSequenceItem { previous_iteration?: PreviousIteration2; key: Key12; - dag_id: DagId18; + dag_id: DagId19; run_id: RunId11; task_id: TaskId6; offset: Offset1; - type?: Type52; + type?: Type53; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1582,14 +1614,14 @@ export interface GetXComSequenceItem { export interface GetXComSequenceSlice { previous_iteration?: PreviousIteration3; key: Key13; - dag_id: DagId19; + dag_id: DagId20; run_id: RunId12; task_id: TaskId7; start: Start; stop: Stop; step: Step; include_prior_dates?: IncludePriorDates1; - type?: Type53; + type?: Type54; } /** * Response to CreateHITLDetailPayload request. @@ -1606,7 +1638,7 @@ export interface HITLDetailRequestResult { multiple?: Multiple1; params?: Params1; assigned_users?: AssignedUsers1; - type?: Type54; + type?: Type55; } /** * Response of InactiveAssets requests. @@ -1616,7 +1648,7 @@ export interface HITLDetailRequestResult { */ export interface InactiveAssetsResult { inactive_assets?: InactiveAssets; - type?: Type55; + type?: Type56; } /** * Add a new value to be redacted in task logs. @@ -1627,7 +1659,7 @@ export interface InactiveAssetsResult { export interface MaskSecret { value: JsonValue; name?: Name14; - type?: Type56; + type?: Type57; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1635,7 +1667,7 @@ export interface MaskSecret { */ export interface OKResponse { ok: Ok; - type?: Type57; + type?: Type58; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1646,7 +1678,7 @@ export interface PrevSuccessfulDagRunResult { data_interval_end?: DataIntervalEnd3; start_date?: StartDate4; end_date?: EndDate3; - type?: Type58; + type?: Type59; } /** * Response containing previous Dag run information. @@ -1656,7 +1688,7 @@ export interface PrevSuccessfulDagRunResult { */ export interface PreviousDagRunResult { dag_run?: DagRun | null; - type?: Type59; + type?: Type60; } /** * Schema for response with previous TaskInstance information. @@ -1666,7 +1698,7 @@ export interface PreviousDagRunResult { */ export interface PreviousTIResponse { task_id: TaskId8; - dag_id: DagId20; + dag_id: DagId21; run_id: RunId13; logical_date?: LogicalDate5; start_date?: StartDate5; @@ -1686,7 +1718,7 @@ export interface PreviousTIResponse { */ export interface PreviousTIResult { task_instance?: PreviousTIResponse | null; - type?: Type60; + type?: Type61; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1696,7 +1728,7 @@ export interface PutVariable { key: Key14; value: Value1; description: Description; - type?: Type61; + type?: Type62; } /** * Update a task instance state to reschedule/up_for_reschedule. @@ -1708,14 +1740,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. @@ -1729,14 +1761,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; } /** @@ -1747,7 +1779,7 @@ export interface SetAssetStateStoreByName { name: Name15; key: Key15; value: JsonValue; - type?: Type66; + type?: Type67; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1757,7 +1789,7 @@ export interface SetAssetStateStoreByUri { uri: Uri9; key: Key16; value: JsonValue; - type?: Type67; + type?: Type68; } /** * Tell the supervisor the task's ``execution_timeout`` so it can enforce it from outside the task process. @@ -1767,7 +1799,7 @@ export interface SetAssetStateStoreByUri { */ export interface SetExecutionTimeout { timeout_seconds: TimeoutSeconds; - type?: Type68; + type?: Type69; } /** * Payload for setting RTIF for a task instance. @@ -1777,7 +1809,7 @@ export interface SetExecutionTimeout { */ export interface SetRenderedFields { rendered_fields: RenderedFields; - type?: Type69; + type?: Type70; } export interface RenderedFields { [k: string]: JsonValue; @@ -1790,7 +1822,7 @@ export interface RenderedFields { */ export interface SetRenderedMapIndex { rendered_map_index: RenderedMapIndex3; - type?: Type70; + type?: Type71; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1801,7 +1833,7 @@ export interface SetTaskStateStore { key: Key17; value: JsonValue; expires_at: ExpiresAt; - type?: Type71; + type?: Type72; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1810,13 +1842,13 @@ export interface SetTaskStateStore { export interface SetXCom { key: Key18; value: JsonValue; - dag_id: DagId21; + dag_id: DagId22; run_id: RunId14; task_id: TaskId9; map_index?: MapIndex7; dag_result?: DagResult1; mapped_length?: MappedLength; - type?: Type72; + type?: Type73; } /** * Update state of downstream tasks within a task instance to 'skipped', while updating current task to success state. @@ -1826,7 +1858,7 @@ export interface SetXCom { */ export interface SkipDownstreamTasks { tasks: Tasks; - type?: Type73; + type?: Type74; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1839,7 +1871,7 @@ export interface StartupDetails { start_date: StartDate6; ti_context: TIRunContext; sentry_integration: SentryIntegration; - type?: Type74; + type?: Type75; } /** * Update a task's state to success. Includes task_outlets and outlet_events for registering asset events. @@ -1853,7 +1885,7 @@ export interface SucceedTask { task_outlets?: TaskOutlets; outlet_events?: OutletEvents; rendered_map_index?: RenderedMapIndex4; - type?: Type75; + type?: Type76; } /** * Response containing count of Task Instances matching certain filters. @@ -1863,7 +1895,7 @@ export interface SucceedTask { */ export interface TICount { count: Count1; - type?: Type76; + type?: Type77; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1871,7 +1903,7 @@ export interface TICount { */ export interface TaskBreadcrumbsResult { breadcrumbs: Breadcrumbs; - type?: Type77; + type?: Type78; } /** * Response containing the first reschedule date for a task instance. @@ -1881,7 +1913,7 @@ export interface TaskBreadcrumbsResult { */ export interface TaskRescheduleStartDate { start_date: StartDate7; - type?: Type78; + type?: Type79; } /** * Update a task's state. @@ -1896,7 +1928,7 @@ export interface TaskRescheduleStartDate { export interface TaskState { state: State8; end_date?: EndDate8; - type?: Type79; + type?: Type80; rendered_map_index?: RenderedMapIndex5; retry_reason?: RetryReason1; } @@ -1908,7 +1940,7 @@ export interface TaskState { */ export interface TaskStateStoreResult { value: JsonValue | null; - type?: Type80; + type?: Type81; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1916,7 +1948,7 @@ export interface TaskStateStoreResult { */ export interface TaskStatesResult { task_states: TaskStates; - type?: Type81; + type?: Type82; } export interface TaskStates { [k: string]: unknown; @@ -1932,9 +1964,9 @@ export interface TriggerDagRun { reset_dag_run?: ResetDagRun; partition_key?: PartitionKey7; note?: Note2; - dag_id: DagId22; + dag_id: DagId23; run_id: DagRunId; - type?: Type82; + type?: Type83; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1943,7 +1975,7 @@ export interface TriggerDagRun { export interface UpdateDagRunNote { ti_id: TiId9; note: Note3; - type?: Type83; + type?: Type84; } /** * Update the response content part of an existing Human-in-the-loop response. @@ -1955,7 +1987,7 @@ export interface UpdateHITLDetail { ti_id: TiId10; chosen_options: ChosenOptions; params_input?: ParamsInput; - type?: Type84; + type?: Type85; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1963,7 +1995,7 @@ export interface UpdateHITLDetail { */ export interface ValidateInletsAndOutlets { ti_id: TiId11; - type?: Type85; + type?: Type86; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1972,7 +2004,7 @@ export interface ValidateInletsAndOutlets { export interface VariableKeysResult { keys: Keys; total_entries: TotalEntries; - type?: Type86; + type?: Type87; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1981,7 +2013,7 @@ export interface VariableKeysResult { export interface VariableResult { key: Key19; value: Value2; - type?: Type87; + type?: Type88; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -1989,7 +2021,7 @@ export interface VariableResult { */ export interface XComCountResponse { len: Len; - type?: Type88; + type?: Type89; } /** * Response to ReadXCom request. @@ -2000,7 +2032,7 @@ export interface XComCountResponse { export interface XComResult { key: Key20; value: JsonValue | null; - type?: Type89; + type?: Type90; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -2008,7 +2040,7 @@ export interface XComResult { */ export interface XComSequenceIndexResult { root: JsonValue; - type?: Type90; + type?: Type91; } /** * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema @@ -2016,7 +2048,7 @@ export interface XComSequenceIndexResult { */ export interface XComSequenceSliceResult { root: Root; - type?: Type91; + type?: Type92; } /** Cadwyn schema version this SDK was generated against. diff --git a/uv.lock b/uv.lock index 5439a258bbed6..c97ff0a5de405 100644 --- a/uv.lock +++ b/uv.lock @@ -9321,6 +9321,7 @@ dependencies = [ { name = "asgiref" }, { name = "attrs" }, { name = "babel" }, + { name = "cadwyn" }, { name = "colorlog" }, { name = "fsspec" }, { name = "greenback" }, @@ -9340,6 +9341,7 @@ dependencies = [ { name = "pydantic" }, { name = "pygtrie" }, { name = "python-dateutil" }, + { name = "pyyaml" }, { name = "structlog" }, { name = "tenacity" }, { name = "typing-extensions" }, @@ -9396,6 +9398,7 @@ requires-dist = [ { name = "asgiref", marker = "python_full_version >= '3.14'", specifier = ">=3.11.1" }, { name = "attrs", specifier = ">=24.2.0,!=25.2.0" }, { name = "babel", specifier = ">=2.17.0" }, + { name = "cadwyn", specifier = ">=6.1.1" }, { name = "colorlog", specifier = ">=6.8.2" }, { name = "datadog", marker = "extra == 'all'", specifier = ">=0.50.0" }, { name = "datadog", marker = "extra == 'datadog'", specifier = ">=0.50.0" }, @@ -9423,6 +9426,7 @@ requires-dist = [ { name = "pydantic", specifier = ">2.11.0" }, { name = "pygtrie", specifier = ">=2.5.0" }, { name = "python-dateutil", specifier = ">=2.7.0" }, + { name = "pyyaml", specifier = ">=6.0.3" }, { name = "sentry-sdk", marker = "extra == 'all'", specifier = ">=2.30.0" }, { name = "sentry-sdk", marker = "extra == 'sentry'", specifier = ">=2.30.0" }, { name = "statsd", marker = "extra == 'all'", specifier = ">=3.3.0" },