Add task instance management commands to airflowctl CLI

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

@@ -96,7 +96,15 @@ def date_param():
     "dags update example_bash_operator --no-is-paused",
     # Dag Run commands
     "dagrun list --dag-id example_bash_operator --state success --limit=1",
-    # XCom commands - need a Dag run with completed tasks
+    # Tasks commands
+    'tasks list example_bash_operator "manual__{date_param}"',
+    'tasks get example_bash_operator "manual__{date_param}" runme_0',
+    "tasks clear example_bash_operator --dry-run",
+    # runme_0 completes as "success" once the triggered run finishes, so updating it
+    # to "success" is rejected with 409 "already in success state". Use "failed" to
+    # exercise a real state transition (valid states: success, failed, skipped).
+    'tasks update example_bash_operator "manual__{date_param}" runme_0 --new-state=failed',
+    # XCom commands - need a DAG run with completed tasks
     'xcom add example_bash_operator "manual__{date_param}" runme_0 {xcom_key} \'{{"test": "value"}}\'',
     'xcom get example_bash_operator "manual__{date_param}" runme_0 {xcom_key}',
     'xcom list example_bash_operator "manual__{date_param}" runme_0',

airflow-ctl/docs/images/command_hashes.txt

@@ -1,4 +1,4 @@
-main:27a22c00dcf32e7a1a4f06672dc8e3c8
+main:164bc97843d5be583c0b48f7a34dc8c8
 assets:6419e20452692f577c4c6f570b74be0c
 auth:d79e9c7d00c432bdbcbc2a86e2e32053
 backfill:74c8737b0a62a86ed3605fa9e6165874
@@ -12,4 +12,5 @@ providers:34502fe09dc0b8b0a13e7e46efdffda6
 variables:f8fc76d3d398b2780f4e97f7cd816646
 version:31f4efdf8de0dbaaa4fac71ff7efecc3
 plugins:4864fd8f356704bd2b3cd1aec3567e35
+tasks:7ab24cac521242b6b6012e2bcd317831
 auth login:9fe2bb1dd5c602beea2eefb33a2b20a8

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

@@ -57,6 +57,7 @@
     PoolsOperations,
     ProvidersOperations,
     ServerResponseError,
+    TasksOperations,
     VariablesOperations,
     VersionOperations,
     XComOperations,
@@ -474,6 +475,12 @@ def plugins(self):
         """Operations related to plugins."""
         return PluginsOperations(self)
 
+    @lru_cache()  # type: ignore[prop-decorator]
+    @property
+    def tasks(self):
+        """Operations related to tasks."""
+        return TasksOperations(self)
+
 
 # API Client Decorator for CLI Actions
 @contextlib.contextmanager

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

@@ -39,6 +39,7 @@
     BulkBodyPoolBody,
     BulkBodyVariableBody,
     BulkResponse,
+    ClearTaskInstancesBody,
     Config,
     ConnectionBody,
     ConnectionCollectionResponse,
@@ -59,6 +60,7 @@
     ImportErrorCollectionResponse,
     ImportErrorResponse,
     JobCollectionResponse,
+    PatchTaskInstanceBody,
     PluginCollectionResponse,
     PluginImportErrorCollectionResponse,
     PoolBody,
@@ -68,6 +70,8 @@
     ProviderCollectionResponse,
     QueuedEventCollectionResponse,
     QueuedEventResponse,
+    TaskInstanceCollectionResponse,
+    TaskInstanceResponse,
     TriggerDAGRunPostBody,
     VariableBody,
     VariableCollectionResponse,
@@ -733,10 +737,16 @@ def delete(self, pool: str) -> str | ServerResponseError:
 
     def update(self, pool_body: PoolPatchBody) -> PoolResponse | ServerResponseError:
         """Update a pool."""
-        try:
-            self.response = self.client.patch(
-                f"pools/{pool_body.pool}", json=pool_body.model_dump(mode="json")
-            )
+        # Workaround: the server's PATCH handler validates the partial body
+        # against ``BasePool`` (see airflow-core/.../services/public/pools.py)
+        # which requires ``include_deferred``. Omitting it fails with
+        # "Field required", sending ``null`` fails with "bool_type". Always
+        # send ``include_deferred`` (defaulting to False when unset) so PATCH
+        # requests are accepted until the server switches to a partial validator.
+        body = pool_body.model_dump(mode="json", exclude_none=True)
+        body.setdefault("include_deferred", False)
+        try:
+            self.response = self.client.patch(f"pools/{pool_body.pool}", json=body)
             return PoolResponse.model_validate_json(self.response.content)
         except ServerResponseError as e:
             raise e
@@ -950,3 +960,45 @@ def list_import_errors(self) -> PluginImportErrorCollectionResponse | ServerResp
             return PluginImportErrorCollectionResponse.model_validate_json(self.response.content)
         except ServerResponseError as e:
             raise e
+
+
+class TasksOperations(BaseOperations):
+    """Tasks operations."""
+
+    def get(self, dag_id: str, dag_run_id: str, task_id: str) -> TaskInstanceResponse:
+        """Get a task instance."""
+        self.response = self.client.get(f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}")
+        return TaskInstanceResponse.model_validate_json(self.response.content)
+
+    def list(self, dag_id: str, dag_run_id: str) -> TaskInstanceCollectionResponse | ServerResponseError:
+        """List task instances."""
+        return super().execute_list(
+            path=f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances",
+            data_model=TaskInstanceCollectionResponse,
+        )
+
+    def clear(
+        self, dag_id: str, body: ClearTaskInstancesBody
+    ) -> TaskInstanceCollectionResponse | ServerResponseError:
+        """Clear task instances."""
+        self.response = self.client.post(
+            f"dags/{dag_id}/clearTaskInstances",
+            json=body.model_dump(mode="json", exclude_unset=True),
+        )
+        return TaskInstanceCollectionResponse.model_validate_json(self.response.content)
+
+    def update(
+        self,
+        dag_id: str,
+        dag_run_id: str,
+        task_id: str,
+        body: PatchTaskInstanceBody,
+        map_index: int | None = None,
+    ) -> TaskInstanceCollectionResponse:
+        """Update task instance state. When map_index is given, only that mapped instance is affected."""
+        if map_index is not None:
+            path = f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}"
+        else:
+            path = f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}"
+        self.response = self.client.patch(path, json=body.model_dump(mode="json", exclude_unset=True))
+        return TaskInstanceCollectionResponse.model_validate_json(self.response.content)