Added dagrun clear command to airflowctl

airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py

@@ -100,6 +100,7 @@ def date_param():
     'xcom list example_bash_operator "manual__{date_param}" runme_0',
     'xcom edit example_bash_operator "manual__{date_param}" runme_0 {xcom_key} \'{{"updated": "value"}}\'',
     'xcom delete example_bash_operator "manual__{date_param}" runme_0 {xcom_key}',
+    'dagrun clear example_bash_operator "manual__{date_param}" --no-dry-run',
     # Jobs commands
     "jobs list",
     # Pools commands

airflow-ctl/docs/images/command_hashes.txt

@@ -5,7 +5,7 @@ backfill:74c8737b0a62a86ed3605fa9e6165874
 config:a3d936cb15fe3b547bf6c82cf93d923f
 connections:942f9f88cb908c28bf5c19159fc5065b
 dags:6b38e6bcd491bc1941e7814b77e63bde
-dagrun:c32e0011aa9a845456c778786717208e
+dagrun:766addfeaa88924043781cd57654bf95
 jobs:a5b644c5da8889443bb40ee10b599270
 pools:19efe105b9515ab1926ebcaf0e028d71
 providers:34502fe09dc0b8b0a13e7e46efdffda6

airflow-ctl/src/airflowctl/api/operations.py

@@ -39,6 +39,7 @@
     BulkBodyPoolBody,
     BulkBodyVariableBody,
     BulkResponse,
+    ClearTaskInstanceCollectionResponse,
     Config,
     ConnectionBody,
     ConnectionCollectionResponse,
@@ -49,6 +50,7 @@
     DAGDetailsResponse,
     DAGPatchBody,
     DAGResponse,
+    DAGRunClearBody,
     DAGRunCollectionResponse,
     DAGRunResponse,
     DagStatsCollectionResponse,
@@ -645,6 +647,27 @@ def list(
         except ServerResponseError as e:
             raise e
 
+    def clear(
+        self, dag_id: str, dag_run_id: str, dry_run: bool | None = None, only_failed: bool | None = None
+    ) -> ClearTaskInstanceCollectionResponse | DAGRunResponse | ServerResponseError:
+        """Clear a Dag run."""
+        dry_run = dry_run if dry_run is not None else True
+        only_failed = only_failed if only_failed is not None else False
+        body = DAGRunClearBody(
+            dry_run=dry_run,
+            only_failed=only_failed,
+        )
+        try:
+            self.response = self.client.post(
+                f"/dags/{dag_id}/dagRuns/{dag_run_id}/clear",
+                json=body.model_dump(mode="json"),
+            )
+            if dry_run:
+                return ClearTaskInstanceCollectionResponse.model_validate_json(self.response.content)
+            return DAGRunResponse.model_validate_json(self.response.content)
+        except ServerResponseError as e:
+            raise e
+
 
 class JobsOperations(BaseOperations):
     """Job operations."""

airflow-ctl/src/airflowctl/ctl/cli_config.py

@@ -394,7 +394,17 @@ def __init__(self, file_path: str | Path | None = None):
         # Exclude parameters that are not needed for CLI from datamodels
         self.excluded_parameters = ["schema_"]
         # This list is used to determine if the command/operation needs to output data
-        self.output_command_list = ["list", "get", "create", "delete", "update", "trigger", "add", "edit"]
+        self.output_command_list = [
+            "list",
+            "get",
+            "create",
+            "delete",
+            "update",
+            "trigger",
+            "add",
+            "edit",
+            "clear",
+        ]
         self.exclude_operation_names = ["LoginOperations", "VersionOperations", "BaseOperations"]
         self.exclude_method_names = [
             "error",

airflow-ctl/src/airflowctl/ctl/help_texts.yaml

@@ -69,6 +69,7 @@ dags:
 dagrun:
   get: "Retrieve a Dag run by Dag ID and run ID"
   list: "List Dag runs, optionally filtered by state and date range"
+  clear: "Clear a Dag run by Dag ID and run ID"
 
 jobs:
   list: "List scheduler, triggerer, and other Airflow jobs"

airflow-ctl/tests/airflow_ctl/api/test_operations.py

@@ -48,6 +48,7 @@
     BulkCreateActionPoolBody,
     BulkCreateActionVariableBody,
     BulkResponse,
+    ClearTaskInstanceCollectionResponse,
     Config,
     ConfigOption,
     ConfigSection,
@@ -91,6 +92,7 @@
     QueuedEventCollectionResponse,
     QueuedEventResponse,
     ReprocessBehavior,
+    TaskInstanceResponse,
     TriggerDAGRunPostBody,
     VariableBody,
     VariableCollectionResponse,
@@ -1172,6 +1174,26 @@ class TestDagRunOperations:
             )
         ],
     )
+    clear_task_instance_collection_response = ClearTaskInstanceCollectionResponse(
+        task_instances=[
+            TaskInstanceResponse(
+                id=uuid.uuid4(),
+                task_id="task_1",
+                dag_id="test_dag",
+                dag_run_id="manual__2024-12-31T23:59:59+00:00",
+                map_index=-1,
+                run_after=datetime.datetime(2024, 12, 31, 23, 59, 59),
+                try_number=1,
+                max_tries=0,
+                task_display_name="task_1",
+                dag_display_name="test_dag",
+                pool="default_pool",
+                pool_slots=1,
+                executor_config="{}",
+            )
+        ],
+        total_entries=1,
+    )
 
     dag_run_collection_response = DAGRunCollectionResponse(
         dag_runs=[dag_run_response],
@@ -1273,6 +1295,30 @@ def handle_request(request: httpx.Request) -> httpx.Response:
         assert "state" not in captured_params
         assert captured_params["limit"] == "5"
 
+    def test_clear(self):
+        def handle_request(request: httpx.Request) -> httpx.Response:
+            assert request.url.path == f"/api/v2/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/clear"
+            request_body = json.loads(request.content)
+            assert request_body["dry_run"] is False
+            return httpx.Response(200, json=json.loads(self.dag_run_response.model_dump_json()))
+
+        client = make_api_client(transport=httpx.MockTransport(handle_request))
+        response = client.dag_runs.clear(dag_id=self.dag_id, dag_run_id=self.dag_run_id, dry_run=False)
+        assert response == self.dag_run_response
+
+    def test_clear_dry_run(self):
+        def handle_request(request: httpx.Request) -> httpx.Response:
+            assert request.url.path == f"/api/v2/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/clear"
+            request_body = json.loads(request.content)
+            assert request_body["dry_run"] is True
+            return httpx.Response(
+                200, json=json.loads(self.clear_task_instance_collection_response.model_dump_json())
+            )
+
+        client = make_api_client(transport=httpx.MockTransport(handle_request))
+        response = client.dag_runs.clear(dag_id=self.dag_id, dag_run_id=self.dag_run_id, dry_run=True)
+        assert response == self.clear_task_instance_collection_response
+
 
 class TestJobsOperations:
     job_response = JobResponse(