diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py index 4ea8a3c1f7297..84bbcf4669d8f 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py @@ -227,6 +227,12 @@ class ClearTaskInstancesBody(StrictBaseModel): ) prevent_running_task: bool = False note: Annotated[str, StringConstraints(max_length=1000)] | None = None + include_downstream_dags: bool = Field( + default=False, + description="If True, also clear tasks in downstream Dags that are linked via " + "ExternalTaskMarker. Follows transitive dependencies up to the recursion_depth " + "configured on each ExternalTaskMarker.", + ) @model_validator(mode="before") @classmethod diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index fea2773828ced..028c86411028e 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -9076,6 +9076,12 @@ paths: schema: $ref: '#/components/schemas/HTTPExceptionResponse' description: Forbidden + '400': + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPExceptionResponse' + description: Bad Request '404': content: application/json: @@ -12525,6 +12531,13 @@ components: maxLength: 1000 - type: 'null' title: Note + include_downstream_dags: + type: boolean + title: Include Downstream Dags + description: If True, also clear tasks in downstream Dags that are linked + via ExternalTaskMarker. Follows transitive dependencies up to the recursion_depth + configured on each ExternalTaskMarker. + default: false additionalProperties: false type: object title: ClearTaskInstancesBody diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py index 214179cdca9a2..620e046a6dda8 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py @@ -22,11 +22,13 @@ import structlog from fastapi import Depends, HTTPException, Query, status +from pendulum.parsing.exceptions import ParserError from sqlalchemy import or_, select from sqlalchemy.orm import joinedload from sqlalchemy.sql.selectable import Select -from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity +from airflow.api_fastapi.app import get_auth_manager +from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity, DagDetails from airflow.api_fastapi.common.cursors import ( apply_cursor_filter, encode_cursor, @@ -111,10 +113,11 @@ _reload_tis_with_rendered_fields, ) from airflow.api_fastapi.logging.decorators import action_logging -from airflow.exceptions import AirflowClearRunningTaskException, TaskNotFound -from airflow.models import Base, DagRun +from airflow.exceptions import AirflowClearRunningTaskException, DagNotFound, TaskNotFound +from airflow.models import Base, DagModel, DagRun from airflow.models.taskinstance import TaskInstance as TI, clear_task_instances from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH +from airflow.serialization.definitions.dag import MaxRecursionDepthError from airflow.ti_deps.dep_context import DepContext from airflow.ti_deps.dependencies_deps import SCHEDULER_QUEUED_DEPS from airflow.utils.db import get_query_count @@ -839,7 +842,9 @@ def get_mapped_task_instance_try_details( @task_instances_router.post( "/clearTaskInstances", - responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT]), + responses=create_openapi_http_exception_doc( + [status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT] + ), dependencies=[ Depends(action_logging()), Depends(requires_access_dag(method="PUT", access_entity=DagAccessEntity.TASK_INSTANCE)), @@ -933,30 +938,69 @@ def _collect_relatives(run_id: str, direction: Literal["upstream", "downstream"] *((t, m) for t, m in mapped_tasks_tuples if t not in normal_task_ids), ] + # Follow ExternalTaskMarker connections when explicitly requested via include_downstream_dags, or + # automatically whenever downstream clearing is selected (restoring Airflow 2 behavior) + include_dependent_dags = body.include_downstream_dags or downstream + task_instances: Sequence[TI] - if dag_run_id is not None and not (past or future): - # Use run_id-based clearing when we have a specific dag_run_id and not using past/future - task_instances = dag.clear( - dry_run=True, - task_ids=task_markers_to_clear, - run_id=dag_run_id, - session=session, - run_on_latest_version=resolved_run_on_latest, - only_failed=body.only_failed, - only_running=body.only_running, - ) - else: - # Use date-based clearing when no dag_run_id or when past/future is specified - task_instances = dag.clear( - dry_run=True, - task_ids=task_markers_to_clear, - start_date=body.start_date, - end_date=body.end_date, - session=session, - run_on_latest_version=resolved_run_on_latest, - only_failed=body.only_failed, - only_running=body.only_running, - ) + try: + if dag_run_id is not None and not (past or future): + # Use run_id-based clearing when we have a specific dag_run_id and not using past/future + task_instances = dag.clear( + dry_run=True, + task_ids=task_markers_to_clear, + run_id=dag_run_id, + session=session, + run_on_latest_version=resolved_run_on_latest, + only_failed=body.only_failed, + only_running=body.only_running, + include_dependent_dags=include_dependent_dags, + dag_bag=dag_bag, + ) + else: + # Use date-based clearing when no dag_run_id or when past/future is specified + task_instances = dag.clear( + dry_run=True, + task_ids=task_markers_to_clear, + start_date=body.start_date, + end_date=body.end_date, + session=session, + run_on_latest_version=resolved_run_on_latest, + only_failed=body.only_failed, + only_running=body.only_running, + include_dependent_dags=include_dependent_dags, + dag_bag=dag_bag, + ) + + except MaxRecursionDepthError as e: + raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) from e + except ParserError as e: + raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid logical_date: {e}") from e + except DagNotFound as e: + raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e + + if include_dependent_dags: + # Ensure proper access to downstream Dags/tasks with dag.clear and include_dependent_dags + auth_manager = get_auth_manager() + all_dag_ids = {ti.dag_id for ti in task_instances} # Retrieve all Dag ID's from task instances + + # Used to find a team name from a Dag ID + dag_id_to_team = DagModel.get_dag_id_to_team_name_mapping(list(all_dag_ids), session=session) + + # set of Dag ID's that can be cleared + editable_dag_ids = { + dependent_dag_id + for dependent_dag_id in all_dag_ids + if auth_manager.is_authorized_dag( + method="PUT", + access_entity=DagAccessEntity.TASK_INSTANCE, + details=DagDetails(id=dependent_dag_id, team_name=dag_id_to_team.get(dependent_dag_id)), + user=user, + ) + } + + # list of all TI's that can be cleared (TI's within the Dags from above) + task_instances = [ti for ti in task_instances if ti.dag_id in editable_dag_ids] if not dry_run: try: diff --git a/airflow-core/src/airflow/models/dagbag.py b/airflow-core/src/airflow/models/dagbag.py index c4bd8eceea102..65613a366a8ca 100644 --- a/airflow-core/src/airflow/models/dagbag.py +++ b/airflow-core/src/airflow/models/dagbag.py @@ -235,11 +235,24 @@ def iter_all_latest_version_dags(self, *, session: Session) -> Generator[Seriali yield dag def get_latest_version_of_dag(self, dag_id: str, *, session: Session) -> SerializedDAG | None: - """Get the latest version of a dag by its id.""" + """Get the latest version of a dag by its id, using cache if enabled.""" from airflow.models.serialized_dag import SerializedDagModel if not (serdag := SerializedDagModel.get(dag_id, session=session)): return None + + with self._lock: + cached = self._dags.get(serdag.dag_version_id) + + # Dag exists in cache and the cached/serialized Dag hashes match, return cached Dag + if cached is not None and cached.dag_hash == serdag.dag_hash: + if self._use_cache: + stats.incr("api_server.dag_bag.cache_hit") + return cached.dag + + if self._use_cache: + stats.incr("api_server.dag_bag.cache_miss") + return self._read_dag(serdag) diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py b/airflow-core/src/airflow/serialization/definitions/dag.py index 8ed0fee2ccabd..87d624b55f1bf 100644 --- a/airflow-core/src/airflow/serialization/definitions/dag.py +++ b/airflow-core/src/airflow/serialization/definitions/dag.py @@ -27,6 +27,7 @@ from typing import TYPE_CHECKING, TypedDict, cast, overload import attrs +import pendulum import structlog from sqlalchemy import func, or_, select, tuple_ @@ -35,6 +36,7 @@ from airflow.configuration import conf as airflow_conf from airflow.exceptions import ( AirflowException, + DagNotFound, DagNotPartitionedError, DagVersionNotFound, InvalidPartitionKeyError, @@ -44,10 +46,12 @@ from airflow.models.base import ID_LEN from airflow.models.dag import DagModel from airflow.models.dag_version import DagVersion +from airflow.models.dagbag import DBDagBag from airflow.models.dagbundle import DagBundleModel from airflow.models.dagrun import DagRun from airflow.models.deadline import Deadline from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel +from airflow.models.renderedtifields import RenderedTaskInstanceFields from airflow.models.taskinstancekey import TaskInstanceKey from airflow.models.tasklog import LogTemplate from airflow.sdk.definitions.deadline import VariableInterval @@ -92,6 +96,10 @@ class EdgeInfoType(TypedDict): label: str | None +class MaxRecursionDepthError(AirflowException): + """Raised when max recursion depth exceeded.""" + + @attrs.define(eq=False, hash=False, slots=False) class SerializedDAG: """ @@ -996,9 +1004,11 @@ def _get_task_instances( end_date: datetime.datetime | None, run_id: str | None, state: TaskInstanceState | Sequence[TaskInstanceState], + include_dependent_dags: bool = ..., exclude_task_ids: Collection[str | tuple[str, int]] | None, exclude_run_ids: frozenset[str] | None, session: Session, + dag_bag: DBDagBag | None = ..., ) -> Iterable[TaskInstance]: ... # pragma: no cover @overload @@ -1011,9 +1021,14 @@ def _get_task_instances( end_date: datetime.datetime | None, run_id: str | None, state: TaskInstanceState | Sequence[TaskInstanceState], + include_dependent_dags: bool = ..., exclude_task_ids: Collection[str | tuple[str, int]] | None, exclude_run_ids: frozenset[str] | None, session: Session, + dag_bag: DBDagBag | None = ..., + recursion_depth: int = ..., + max_recursion_depth: int = ..., + visited_external_tis: set[tuple[str, str, str, int]] = ..., ) -> set[TaskInstanceKey]: ... # pragma: no cover def _get_task_instances( @@ -1025,9 +1040,14 @@ def _get_task_instances( end_date: datetime.datetime | None, run_id: str | None, state: TaskInstanceState | Sequence[TaskInstanceState], + include_dependent_dags: bool = False, exclude_task_ids: Collection[str | tuple[str, int]] | None, exclude_run_ids: frozenset[str] | None, session: Session, + dag_bag: DBDagBag | None = None, + recursion_depth: int = 0, + max_recursion_depth: int | None = None, + visited_external_tis: set[tuple[str, str, str, int]] | None = None, ) -> Iterable[TaskInstance] | set[TaskInstanceKey]: from airflow.models.taskinstance import TaskInstance @@ -1104,6 +1124,106 @@ def apply_state_filter(query): else: tis_full = apply_state_filter(tis_full) + if include_dependent_dags: + # Recursively find external tasks indicated by ExternalTaskMarker + from airflow.providers.standard.sensors.external_task import ExternalTaskMarker + + # Build a full-object query for identifying ExternalTaskMarker TIs in the current set + if as_pk_tuple: + all_ti_rows = session.execute(tis_pk).all() + condition = TaskInstance.filter_for_tis( + TaskInstanceKey(**cols._mapping) for cols in all_ti_rows + ) + marker_query = select(TaskInstance).where(condition) if condition is not None else None + else: + marker_query = tis_full + + if marker_query is not None: + if visited_external_tis is None: + visited_external_tis = set() + + external_marker_tis = session.scalars( + marker_query.where(TaskInstance.operator == ExternalTaskMarker.__name__) + ) + + for ti in external_marker_tis: + if (ti_key := ti.key.primary) in visited_external_tis: + continue + + visited_external_tis.add(ti_key) + task: ExternalTaskMarker = cast("ExternalTaskMarker", self.get_task(ti.task_id)) + + if max_recursion_depth is None: + # Maximum recursion depth is set from the first ExternalTaskMarker encountered + max_recursion_depth = task.recursion_depth + + if recursion_depth + 1 > max_recursion_depth: + raise MaxRecursionDepthError( + f"Maximum recursion depth {max_recursion_depth} reached for " + f"{ExternalTaskMarker.__name__} {ti.task_id}. " + f"Attempted to clear too many tasks or there may be a cyclic dependency." + ) + + if ti.dag_run.logical_date is None: + continue + + # Retrieve the logical date from the TI, Dag/task ID's from the task + logical_date_str: str = ti.dag_run.logical_date.isoformat() + external_dag_id = task.external_dag_id + external_task_id = task.external_task_id + + if rendered := RenderedTaskInstanceFields.get_templated_fields(ti, session=session): + external_dag_id = rendered.get("external_dag_id", external_dag_id) + external_task_id = rendered.get("external_task_id", external_task_id) + + if "logical_date" in rendered: + logical_date_str = rendered["logical_date"] + + external_logical_date = pendulum.parse(logical_date_str) + external_tis = session.scalars( + select(TaskInstance) + .join(TaskInstance.dag_run) + .where( + TaskInstance.dag_id == external_dag_id, + TaskInstance.task_id == external_task_id, + DagRun.logical_date == external_logical_date, + ) + ) + + # Load the DagBag such that Dags can be extracted from it + if not dag_bag: + dag_bag = DBDagBag(load_op_links=False) + + for tii in external_tis: + external_dag = dag_bag.get_latest_version_of_dag(tii.dag_id, session=session) + if not external_dag: + raise DagNotFound(f"Could not find Dag {tii.dag_id}") + + downstream = external_dag.partial_subset( + task_ids=[tii.task_id], + include_upstream=False, + include_downstream=True, + ) + + result.update( + downstream._get_task_instances( + task_ids=None, + run_id=tii.run_id, + start_date=None, + end_date=None, + state=state, + include_dependent_dags=include_dependent_dags, + as_pk_tuple=True, + exclude_task_ids=exclude_task_ids, + exclude_run_ids=exclude_run_ids, + session=session, + dag_bag=dag_bag, + recursion_depth=recursion_depth + 1, + max_recursion_depth=max_recursion_depth, + visited_external_tis=visited_external_tis, + ) + ) + if result or as_pk_tuple: # Only execute the `ti` query if we have also collected some other results if as_pk_tuple: @@ -1154,6 +1274,8 @@ def clear( exclude_task_ids: frozenset[str] | frozenset[tuple[str, int]] | None = frozenset(), exclude_run_ids: frozenset[str] | None = frozenset(), run_on_latest_version: bool = False, + include_dependent_dags: bool = False, + dag_bag: DBDagBag | None = None, ) -> set[str]: ... # pragma: no cover @overload @@ -1171,6 +1293,8 @@ def clear( exclude_task_ids: frozenset[str] | frozenset[tuple[str, int]] | None = frozenset(), exclude_run_ids: frozenset[str] | None = frozenset(), run_on_latest_version: bool = False, + include_dependent_dags: bool = False, + dag_bag: DBDagBag | None = None, ) -> list[TaskInstance]: ... # pragma: no cover @overload @@ -1205,6 +1329,8 @@ def clear( exclude_task_ids: frozenset[str] | frozenset[tuple[str, int]] | None = frozenset(), exclude_run_ids: frozenset[str] | None = frozenset(), run_on_latest_version: bool = False, + include_dependent_dags: bool = False, + dag_bag: DBDagBag | None = None, ) -> int: ... # pragma: no cover @overload @@ -1222,6 +1348,8 @@ def clear( exclude_task_ids: frozenset[str] | frozenset[tuple[str, int]] | None = frozenset(), exclude_run_ids: frozenset[str] | None = frozenset(), run_on_latest_version: bool = False, + include_dependent_dags: bool = False, + dag_bag: DBDagBag | None = None, ) -> list[TaskInstance]: ... # pragma: no cover @overload @@ -1239,6 +1367,8 @@ def clear( exclude_task_ids: frozenset[str] | frozenset[tuple[str, int]] | None = frozenset(), exclude_run_ids: frozenset[str] | None = frozenset(), run_on_latest_version: bool = False, + include_dependent_dags: bool = False, + dag_bag: DBDagBag | None = None, ) -> int: ... # pragma: no cover @provide_session @@ -1258,6 +1388,8 @@ def clear( exclude_task_ids: frozenset[str] | frozenset[tuple[str, int]] | None = frozenset(), exclude_run_ids: frozenset[str] | None = frozenset(), run_on_latest_version: bool = False, + include_dependent_dags: bool = False, + dag_bag: DBDagBag | None = None, ) -> int | Iterable[TaskInstance] | set[str]: """ Clear a set of task instances associated with the current dag for a specified date range. @@ -1277,6 +1409,12 @@ def clear( :param exclude_task_ids: A set of ``task_id`` or (``task_id``, ``map_index``) tuples that should not be cleared :param exclude_run_ids: A set of ``run_id`` or (``run_id``) + :param include_dependent_dags: If True, also clear tasks in downstream Dags that are + linked via ExternalTaskMarker. Follows transitive dependencies up to the + ``recursion_depth`` configured on each ExternalTaskMarker. + :param dag_bag: An existing ``DBDagBag`` to reuse (e.g. the request-scoped bag) when + resolving downstream dags for ``include_dependent_dags``, instead of creating a new, + uncached, un-configured one. """ from airflow.models.taskinstance import ( _get_new_task_ids, @@ -1329,9 +1467,11 @@ def clear( end_date=end_date, run_id=run_id, state=state, + include_dependent_dags=include_dependent_dags, session=session, exclude_task_ids=exclude_task_ids, exclude_run_ids=exclude_run_ids, + dag_bag=dag_bag, ) if dry_run: diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index d14b118651c30..fc5d660bbc121 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -2259,6 +2259,12 @@ export const $ClearTaskInstancesBody = { } ], title: 'Note' + }, + include_downstream_dags: { + type: 'boolean', + title: 'Include Downstream Dags', + description: 'If True, also clear tasks in downstream Dags that are linked via ExternalTaskMarker. Follows transitive dependencies up to the recursion_depth configured on each ExternalTaskMarker.', + default: false } }, additionalProperties: false, diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts index c31f14f957b33..44add77c077c5 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts @@ -2976,6 +2976,7 @@ export class TaskInstanceService { body: data.requestBody, mediaType: 'application/json', errors: { + 400: 'Bad Request', 401: 'Unauthorized', 403: 'Forbidden', 404: 'Not Found', diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index 2c6a870e0e323..2d2989bb38ef4 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -709,6 +709,10 @@ export type ClearTaskInstancesBody = { run_on_latest_version?: boolean | null; prevent_running_task?: boolean; note?: string | null; + /** + * If True, also clear tasks in downstream Dags that are linked via ExternalTaskMarker. Follows transitive dependencies up to the recursion_depth configured on each ExternalTaskMarker. + */ + include_downstream_dags?: boolean; }; /** @@ -7045,6 +7049,10 @@ export type $OpenApiTs = { * Successful Response */ 200: TaskInstanceCollectionResponse; + /** + * Bad Request + */ + 400: HTTPExceptionResponse; /** * Unauthorized */ diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py index f68acf75d60e0..62e074afd16c6 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py @@ -21,6 +21,7 @@ import itertools import math import os +import uuid from datetime import timedelta from typing import TYPE_CHECKING, Any from unittest import mock @@ -28,14 +29,17 @@ import pendulum import pytest from fastapi.testclient import TestClient +from pendulum.parsing.exceptions import ParserError from sqlalchemy import delete, func, select, update from sqlalchemy.orm import joinedload from airflow._shared.state import TaskScope from airflow._shared.timezones.timezone import datetime +from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser from airflow.dag_processing.bundles.manager import DagBundlesManager from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db +from airflow.exceptions import DagNotFound from airflow.jobs.job import Job from airflow.jobs.triggerer_job_runner import TriggererJobRunner from airflow.models import DagModel, DagRun, Log, TaskInstance @@ -49,6 +53,7 @@ from airflow.models.team import Team from airflow.models.trigger import Trigger from airflow.sdk import BaseOperator +from airflow.serialization.definitions.dag import MaxRecursionDepthError from airflow.state.metastore import MetastoreBackend from airflow.utils.platform import getuser from airflow.utils.state import DagRunState, State, TaskInstanceState @@ -3733,6 +3738,135 @@ def test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s assert dagrun.state == "running" assert all(ti.state == "running" for ti in tis) + @pytest.mark.parametrize( + ("post_body", "include_downstream_dags"), + [ + ({"dry_run": True, "only_failed": False, "include_downstream_dags": True}, True), + ({"dry_run": True, "only_failed": False, "include_downstream": True}, True), + ({"dry_run": True, "only_failed": False}, False), + ( + { + "dry_run": True, + "only_failed": False, + "dag_run_id": "TEST_DAG_RUN_ID", + "include_downstream_dags": True, + }, + True, + ), + ], + ) + @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", return_value=[]) + def test_include_downstream(self, mock_clear, post_body, include_downstream_dags, test_client, session): + self.create_task_instances(session) + response = test_client.post( + "/dags/example_python_operator/clearTaskInstances", + json=post_body, + ) + assert response.status_code == 200 + assert mock_clear.call_count == 1 + assert mock_clear.call_args.kwargs["include_dependent_dags"] is include_downstream_dags + + @pytest.mark.parametrize( + "post_body", + [ + {"dry_run": True, "only_failed": False, "dag_run_id": "TEST_DAG_RUN_ID"}, + {"dry_run": True, "only_failed": False}, + ], + ) + @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear", return_value=[]) + def test_passes_request_dag_bag_to_clear(self, mock_clear, post_body, test_client, session): + self.create_task_instances(session) + response = test_client.post("/dags/example_python_operator/clearTaskInstances", json=post_body) + + assert response.status_code == 200 + assert mock_clear.call_count == 1 + assert mock_clear.call_args.kwargs["dag_bag"] is test_client.app.state.dag_bag + + @mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances") + @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear") + @mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager") + def test_include_dependent_dags_filters_unauthorized_child_tis( + self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis, test_client, session + ): + """TIs from child Dags the caller cannot edit must be excluded when include_dependent_dags=True.""" + self.create_task_instances(session) + + parent_dag_id = "example_python_operator" + parent_ti = mock.MagicMock(spec=TaskInstance) + parent_ti.dag_id = parent_dag_id + parent_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000001") + + child_dag_id = "child_dag_caller_cannot_edit" + child_ti = mock.MagicMock(spec=TaskInstance) + child_ti.dag_id = child_dag_id + child_ti.id = uuid.UUID("00000000-0000-0000-0000-000000000002") + + mock_dag_clear.return_value = [parent_ti, child_ti] + mock_get_auth_manager.return_value.is_authorized_dag.side_effect = lambda **kwargs: ( + kwargs["details"].id == parent_dag_id + ) + + response = test_client.post( + f"/dags/{parent_dag_id}/clearTaskInstances", + json={"dry_run": False, "include_downstream_dags": True}, + ) + + assert response.status_code == 200 + + auth_calls = mock_get_auth_manager.return_value.is_authorized_dag.call_args_list + + # Auth calls should be made for both the parent and child DAGs + assert {call.kwargs["details"].id for call in auth_calls} == {parent_dag_id, child_dag_id} + + # Auth calls should be for a TI + for call in auth_calls: + assert call.kwargs["method"] == "PUT" + assert call.kwargs["access_entity"] == DagAccessEntity.TASK_INSTANCE + + cleared_tis = mock_clear_tis.call_args[0][0] + assert cleared_tis == [parent_ti] # Child ID's are NOT cleared + + @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear") + def test_cyclic_external_task_marker_returns_400(self, mock_clear, test_client, session): + """A cyclic or too-deep ExternalTaskMarker chain must return 400, not 500.""" + self.create_task_instances(session) + mock_clear.side_effect = MaxRecursionDepthError( + "Maximum recursion depth 1 reached for ExternalTaskMarker marker_task." + ) + response = test_client.post( + "/dags/example_python_operator/clearTaskInstances", + json={"dry_run": True, "include_downstream_dags": True}, + ) + + assert response.status_code == 400 + assert "Maximum recursion depth" in response.json()["detail"] + + @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear") + def test_missing_child_dag_returns_404(self, mock_clear, test_client, session): + """A missing child Dag referenced by ExternalTaskMarker must return 404, not 500.""" + self.create_task_instances(session) + mock_clear.side_effect = DagNotFound("Could not find Dag child_dag") + response = test_client.post( + "/dags/example_python_operator/clearTaskInstances", + json={"dry_run": True, "include_downstream_dags": True}, + ) + + assert response.status_code == 404 + assert response.json()["detail"] == "Could not find Dag child_dag" + + @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear") + def test_invalid_external_task_marker_logical_date_returns_400(self, mock_clear, test_client, session): + """A non-ISO logical_date rendered from an ExternalTaskMarker template must return 400, not 500.""" + self.create_task_instances(session) + mock_clear.side_effect = ParserError("Unable to parse string [not-a-date]") + response = test_client.post( + "/dags/example_python_operator/clearTaskInstances", + json={"dry_run": True, "include_downstream_dags": True}, + ) + + assert response.status_code == 400 + assert "Invalid logical_date" in response.json()["detail"] + def test_should_respond_200_with_reset_dag_run(self, test_client, session): dag_id = "example_python_operator" payload = { diff --git a/airflow-core/tests/unit/models/test_dagbag.py b/airflow-core/tests/unit/models/test_dagbag.py index 79668d4fe54f4..e6a691ac64e90 100644 --- a/airflow-core/tests/unit/models/test_dagbag.py +++ b/airflow-core/tests/unit/models/test_dagbag.py @@ -182,6 +182,65 @@ def test_get_dag_returns_none_when_not_found(self): assert result is None + @patch("airflow.models.serialized_dag.SerializedDagModel.get") + def test_get_latest_version_of_dag_serves_from_cache_on_hash_match(self, mock_get): + """A cached entry whose hash matches the current row is served without re-deserializing.""" + mock_dag = MagicMock(spec=SerializedDAG) + self.db_dag_bag._dags["v1"] = _CacheEntry(mock_dag, "hash1", time.monotonic()) + mock_serdag = MagicMock(spec=SerializedDagModel) + mock_serdag.dag_version_id = "v1" + mock_serdag.dag_hash = "hash1" + mock_get.return_value = mock_serdag + + with patch.object(self.db_dag_bag, "_read_dag") as mock_read_dag: + result = self.db_dag_bag.get_latest_version_of_dag("some_dag", session=self.session) + + assert result == mock_dag + mock_read_dag.assert_not_called() + + @patch("airflow.models.serialized_dag.SerializedDagModel.get") + def test_get_latest_version_of_dag_reloads_on_hash_mismatch(self, mock_get): + """A cached entry with a stale hash (dag updated in place) triggers a fresh deserialize.""" + stale_dag = MagicMock(spec=SerializedDAG) + fresh_dag = MagicMock(spec=SerializedDAG) + self.db_dag_bag._dags["v1"] = _CacheEntry(stale_dag, "old_hash", time.monotonic()) + mock_serdag = MagicMock(spec=SerializedDagModel) + mock_serdag.dag_version_id = "v1" + mock_serdag.dag_hash = "new_hash" + mock_serdag.dag = fresh_dag + mock_get.return_value = mock_serdag + + result = self.db_dag_bag.get_latest_version_of_dag("some_dag", session=self.session) + + assert result == fresh_dag + entry = self.db_dag_bag._dags["v1"] + assert (entry.dag, entry.dag_hash) == (fresh_dag, "new_hash") + + @patch("airflow.models.serialized_dag.SerializedDagModel.get") + def test_get_latest_version_of_dag_deserializes_on_miss(self, mock_get): + """No cache entry for the version deserializes and caches it.""" + fresh_dag = MagicMock(spec=SerializedDAG) + mock_serdag = MagicMock(spec=SerializedDagModel) + mock_serdag.dag_version_id = "v1" + mock_serdag.dag_hash = "hash1" + mock_serdag.dag = fresh_dag + mock_get.return_value = mock_serdag + + result = self.db_dag_bag.get_latest_version_of_dag("some_dag", session=self.session) + + assert result == fresh_dag + entry = self.db_dag_bag._dags["v1"] + assert (entry.dag, entry.dag_hash) == (fresh_dag, "hash1") + + @patch("airflow.models.serialized_dag.SerializedDagModel.get") + def test_get_latest_version_of_dag_returns_none_when_not_found(self, mock_get): + """It should return None if no serialized dag row exists for the dag_id.""" + mock_get.return_value = None + + result = self.db_dag_bag.get_latest_version_of_dag("missing_dag", session=self.session) + + assert result is None + def test_get_dag_reflects_in_place_version_update_end_to_end(self): """End-to-end regression: an in-place version update must be re-read, not served stale. @@ -407,6 +466,36 @@ def test_cache_hit_metric_emitted(self, mock_stats): mock_stats.incr.assert_called_with("api_server.dag_bag.cache_hit") + @patch("airflow.models.serialized_dag.SerializedDagModel.get") + @patch("airflow.models.dagbag.stats") + def test_get_latest_version_of_dag_cache_hit_metric_emitted(self, mock_stats, mock_get): + """A cached, still-current version served by get_latest_version_of_dag counts as a hit.""" + dag_bag = DBDagBag(cache_size=10, cache_ttl=60) + dag_bag._dags["test_version"] = _CacheEntry(MagicMock(), "hash1", time.monotonic()) + mock_serdag = MagicMock(spec=SerializedDagModel) + mock_serdag.dag_version_id = "test_version" + mock_serdag.dag_hash = "hash1" + mock_get.return_value = mock_serdag + + dag_bag.get_latest_version_of_dag("some_dag", session=MagicMock()) + + mock_stats.incr.assert_called_with("api_server.dag_bag.cache_hit") + + @patch("airflow.models.serialized_dag.SerializedDagModel.get") + @patch("airflow.models.dagbag.stats") + def test_get_latest_version_of_dag_cache_miss_metric_emitted(self, mock_stats, mock_get): + """An uncached version deserialized by get_latest_version_of_dag counts as a miss.""" + dag_bag = DBDagBag(cache_size=10, cache_ttl=60) + mock_serdag = MagicMock(spec=SerializedDagModel) + mock_serdag.dag_version_id = "uncached_version" + mock_serdag.dag_hash = "hash1" + mock_serdag.dag = MagicMock(spec=SerializedDAG) + mock_get.return_value = mock_serdag + + dag_bag.get_latest_version_of_dag("some_dag", session=MagicMock()) + + mock_stats.incr.assert_any_call("api_server.dag_bag.cache_miss") + @patch("airflow.models.dagbag.stats") def test_cache_miss_metric_emitted(self, mock_stats): """Test that cache miss metric is emitted when DAG is found in DB but not in cache.""" diff --git a/airflow-core/tests/unit/serialization/definitions/test_dag.py b/airflow-core/tests/unit/serialization/definitions/test_dag.py new file mode 100644 index 0000000000000..2d9815613c07d --- /dev/null +++ b/airflow-core/tests/unit/serialization/definitions/test_dag.py @@ -0,0 +1,391 @@ +# +# 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 unittest import mock + +import pendulum +import pytest + +from airflow.models.dagbag import DBDagBag +from airflow.models.renderedtifields import RenderedTaskInstanceFields +from airflow.providers.standard.sensors.external_task import ExternalTaskMarker, ExternalTaskSensor +from airflow.serialization.definitions.dag import MaxRecursionDepthError + +from tests_common.test_utils.db import ( + clear_db_dags, + clear_db_runs, + clear_db_serialized_dags, + clear_rendered_ti_fields, +) + +pytestmark = pytest.mark.db_test + +EXTERNAL_LOGICAL_DATE = pendulum.datetime(2024, 1, 1, tz="UTC") + + +@pytest.fixture(autouse=True) +def reset_db(): + clear_db_dags() + clear_db_runs() + clear_db_serialized_dags() + clear_rendered_ti_fields() + + +def test_clear_does_not_follow_external_marker_by_default(dag_maker, session): + """Without include_dependent_dags, ExternalTaskMarker links are not followed.""" + with dag_maker("parent_dag", session=session, schedule=None): + ExternalTaskMarker( + task_id="trigger_child", + external_dag_id="child_dag", + external_task_id="wait_for_parent", + recursion_depth=3, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + serialized_parent = dag_maker.serialized_dag + + with dag_maker("child_dag", session=session, schedule=None): + ExternalTaskSensor( + task_id="wait_for_parent", + external_dag_id="parent_dag", + external_task_id="trigger_child", + poke_interval=5, + ) + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + session.flush() + + result = serialized_parent.clear(dry_run=True, only_failed=False, session=session) + + dag_ids = {ti.dag_id for ti in result} + assert "parent_dag" in dag_ids + assert "child_dag" not in dag_ids + + +def test_clear_follows_external_marker_when_include_dependent_dags_enabled(dag_maker, session): + """With include_dependent_dags=True, clear() follows ExternalTaskMarker links into child Dags.""" + with dag_maker("parent_dag", session=session, schedule=None): + ExternalTaskMarker( + task_id="trigger_child", + external_dag_id="child_dag", + external_task_id="wait_for_parent", + recursion_depth=3, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + serialized_parent = dag_maker.serialized_dag + + with dag_maker("child_dag", session=session, schedule=None): + ExternalTaskSensor( + task_id="wait_for_parent", + external_dag_id="parent_dag", + external_task_id="trigger_child", + poke_interval=5, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + session.flush() + + result = serialized_parent.clear( + dry_run=True, only_failed=False, include_dependent_dags=True, session=session + ) + + dag_ids = {ti.dag_id for ti in result} + task_ids = {ti.task_id for ti in result} + + assert "parent_dag" in dag_ids + assert "child_dag" in dag_ids + assert "wait_for_parent" in task_ids + + +def test_clear_reuses_provided_dag_bag_for_external_dags(dag_maker, session): + """Passing an existing dag_bag into clear() reuses it instead of creating an uncached, un-configured one.""" + with dag_maker("parent_dag", session=session, schedule=None): + ExternalTaskMarker( + task_id="trigger_child", + external_dag_id="child_dag", + external_task_id="wait_for_parent", + recursion_depth=3, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + serialized_parent = dag_maker.serialized_dag + + with dag_maker("child_dag", session=session, schedule=None): + ExternalTaskSensor( + task_id="wait_for_parent", + external_dag_id="parent_dag", + external_task_id="trigger_child", + poke_interval=5, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + session.flush() + + provided_dag_bag = DBDagBag() + + with mock.patch("airflow.serialization.definitions.dag.DBDagBag", wraps=DBDagBag) as mock_dbdagbag_cls: + result = serialized_parent.clear( + dry_run=True, + only_failed=False, + include_dependent_dags=True, + session=session, + dag_bag=provided_dag_bag, + ) + + mock_dbdagbag_cls.assert_not_called() + dag_ids = {ti.dag_id for ti in result} + assert "child_dag" in dag_ids + + +def test_clear_creates_dag_bag_when_none_provided(dag_maker, session): + """Without a caller-provided dag_bag, clear() falls back to creating its own.""" + with dag_maker("parent_dag", session=session, schedule=None): + ExternalTaskMarker( + task_id="trigger_child", + external_dag_id="child_dag", + external_task_id="wait_for_parent", + recursion_depth=3, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + serialized_parent = dag_maker.serialized_dag + + with dag_maker("child_dag", session=session, schedule=None): + ExternalTaskSensor( + task_id="wait_for_parent", + external_dag_id="parent_dag", + external_task_id="trigger_child", + poke_interval=5, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + session.flush() + + with mock.patch("airflow.serialization.definitions.dag.DBDagBag", wraps=DBDagBag) as mock_dbdagbag_cls: + serialized_parent.clear(dry_run=True, only_failed=False, include_dependent_dags=True, session=session) + + mock_dbdagbag_cls.assert_called_once_with(load_op_links=False) + + +def test_clear_dependent_dags_deserializes_child_dag_once_across_multiple_markers(dag_maker, session): + """Multiple ExternalTaskMarkers into the same child dag must not each re-deserialize it.""" + with dag_maker("parent_dag", session=session, schedule=None): + ExternalTaskMarker( + task_id="trigger_child_a", + external_dag_id="child_dag", + external_task_id="wait_for_parent_a", + recursion_depth=3, + ) + ExternalTaskMarker( + task_id="trigger_child_b", + external_dag_id="child_dag", + external_task_id="wait_for_parent_b", + recursion_depth=3, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + serialized_parent = dag_maker.serialized_dag + + with dag_maker("child_dag", session=session, schedule=None): + ExternalTaskSensor( + task_id="wait_for_parent_a", + external_dag_id="parent_dag", + external_task_id="trigger_child_a", + poke_interval=5, + ) + ExternalTaskSensor( + task_id="wait_for_parent_b", + external_dag_id="parent_dag", + external_task_id="trigger_child_b", + poke_interval=5, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + session.flush() + + with mock.patch.object( + DBDagBag, "_read_dag", autospec=True, side_effect=DBDagBag._read_dag + ) as mock_read_dag: + result = serialized_parent.clear( + dry_run=True, only_failed=False, include_dependent_dags=True, session=session + ) + + dag_ids = {ti.dag_id for ti in result} + assert "child_dag" in dag_ids + # Only the first ExternalTaskMarker into child_dag triggers an actual deserialize; the + # second is served from the shared DBDagBag cache instead of re-reading/re-deserializing. + child_dag_reads = [call for call in mock_read_dag.call_args_list if call.args[1].dag_id == "child_dag"] + assert len(child_dag_reads) == 1 + + +def test_clear_raises_when_recursion_depth_exceeded(dag_maker, session): + """MaxRecursionDepthError is raised when the dependency chain depth exceeds recursion_depth.""" + with dag_maker("parent_dag", session=session, schedule=None): + ExternalTaskMarker( + task_id="trigger_child", + external_dag_id="child_dag", + external_task_id="wait_for_parent", + recursion_depth=1, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + serialized_parent = dag_maker.serialized_dag + + with dag_maker("child_dag", session=session, schedule=None): + wait_for_parent = ExternalTaskSensor( + task_id="wait_for_parent", + external_dag_id="parent_dag", + external_task_id="trigger_child", + poke_interval=5, + ) + + trigger_grandchild = ExternalTaskMarker( + task_id="trigger_grandchild", + external_dag_id="grandchild_dag", + external_task_id="wait_for_child", + recursion_depth=1, + ) + + wait_for_parent >> trigger_grandchild + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + + with dag_maker("grandchild_dag", session=session, schedule=None): + ExternalTaskSensor( + task_id="wait_for_child", + external_dag_id="child_dag", + external_task_id="trigger_grandchild", + poke_interval=5, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + session.flush() + + with pytest.raises(MaxRecursionDepthError, match="Maximum recursion depth"): + serialized_parent.clear(dry_run=True, only_failed=False, include_dependent_dags=True, session=session) + + +def test_clear_uses_rendered_fields_for_custom_logical_date_template(dag_maker, session): + """When ExternalTaskMarker has a non-default logical_date template, the rendered value from + RenderedTaskInstanceFields is used to locate the child DagRun.""" + child_logical_date = pendulum.datetime(2024, 1, 2, tz="UTC") + custom_template = "{{ (logical_date + macros.timedelta(days=1)).isoformat() }}" + + with dag_maker("parent_dag", session=session, schedule=None): + ExternalTaskMarker( + task_id="trigger_child", + external_dag_id="child_dag", + external_task_id="wait_for_parent", + logical_date=custom_template, + recursion_depth=3, + ) + + parent_run = dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + serialized_parent = dag_maker.serialized_dag + + # Store the rendered logical_date so the code can resolve the child's DagRun date. + parent_ti = next(ti for ti in parent_run.task_instances if ti.task_id == "trigger_child") + parent_ti.refresh_from_task(serialized_parent.get_task("trigger_child")) + rtif = RenderedTaskInstanceFields( + ti=parent_ti, + render_templates=False, + rendered_fields={"logical_date": child_logical_date.isoformat()}, + ) + session.add(rtif) + session.flush() + + with dag_maker("child_dag", session=session, schedule=None): + ExternalTaskSensor( + task_id="wait_for_parent", + external_dag_id="parent_dag", + external_task_id="trigger_child", + poke_interval=5, + ) + + dag_maker.create_dagrun(logical_date=child_logical_date) + session.flush() + + result = serialized_parent.clear( + dry_run=True, only_failed=False, include_dependent_dags=True, session=session + ) + dag_ids = {ti.dag_id for ti in result} + + assert "child_dag" in dag_ids + + +@pytest.mark.parametrize( + ("marker_kwargs", "rendered_fields"), + [ + pytest.param( + {"external_dag_id": "{{ var.value.child_dag }}", "external_task_id": "wait_for_parent"}, + {"external_dag_id": "child_dag", "external_task_id": "wait_for_parent"}, + id="templated-external_dag_id", + ), + pytest.param( + {"external_dag_id": "child_dag", "external_task_id": "{{ var.value.child_task }}"}, + {"external_dag_id": "child_dag", "external_task_id": "wait_for_parent"}, + id="templated-external_task_id", + ), + ], +) +def test_clear_uses_rendered_fields_for_templated_dag_and_task_id( + dag_maker, session, marker_kwargs, rendered_fields +): + """Use value from RenderedTaskInstanceFields if external_dag_id/external_task_id is a Jinja template.""" + with dag_maker("parent_dag", session=session, schedule=None): + ExternalTaskMarker( + task_id="trigger_child", + recursion_depth=3, + **marker_kwargs, + ) + + # Create a parent DAG run and retrieved the serialized DAG + parent_run = dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + serialized_parent = dag_maker.serialized_dag + + # Create the parent TaskInstances that are to be used to trigger the child + parent_ti = next(ti for ti in parent_run.task_instances if ti.task_id == "trigger_child") + parent_ti.refresh_from_task(serialized_parent.get_task("trigger_child")) + rtif = RenderedTaskInstanceFields( + ti=parent_ti, + render_templates=False, + rendered_fields=rendered_fields, + ) + session.add(rtif) + session.flush() + + with dag_maker("child_dag", session=session, schedule=None): + ExternalTaskSensor( + task_id="wait_for_parent", + external_dag_id="parent_dag", + external_task_id="trigger_child", + poke_interval=5, + ) + + dag_maker.create_dagrun(logical_date=EXTERNAL_LOGICAL_DATE) + session.flush() + + # When clearing the parent DAG run, make sure that the child DAG is in the list of DAGs to clear + result = serialized_parent.clear( + dry_run=True, only_failed=False, include_dependent_dags=True, session=session + ) + dag_ids = {ti.dag_id for ti in result} + + assert "child_dag" in dag_ids diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py index 4708365e846e2..719a4d05e669f 100644 --- a/airflow-core/tests/unit/serialization/test_dag_serialization.py +++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py @@ -524,6 +524,7 @@ class TestStringifiedDAGs: @pytest.fixture(autouse=True) def setup_test_cases(self): + DagSerialization._load_operator_extra_links = True with mock.patch.object(BaseHook, "get_connection") as m: m.return_value = Connection( extra=( diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py index a4b7f60fdaf18..4e5e6bcbcc612 100644 --- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py +++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py @@ -301,6 +301,13 @@ class ClearTaskInstancesBody(BaseModel): ] = None prevent_running_task: Annotated[bool | None, Field(title="Prevent Running Task")] = False note: Annotated[Note | None, Field(title="Note")] = None + include_downstream_dags: Annotated[ + bool | None, + Field( + description="If True, also clear tasks in downstream Dags that are linked via ExternalTaskMarker. Follows transitive dependencies up to the recursion_depth configured on each ExternalTaskMarker.", + title="Include Downstream Dags", + ), + ] = False class Value(RootModel[tuple[str, str]]):