Add airflowctl tasks clear command

airflow-core/src/airflow/cli/commands/task_command.py

@@ -493,6 +493,7 @@ def task_render(args, dag: DAG | None = None) -> None:
         )
 
 
+@deprecated_for_airflowctl("airflowctl tasks clear")
 @cli_utils.action_cli(check_db=False)
 @providers_configuration_loaded
 def task_clear(args) -> None:

airflow-core/tests/unit/cli/commands/test_command_deprecations.py

@@ -76,6 +76,7 @@
     (config_command.get_value, "airflowctl config get"),
     (config_command.show_config, "airflowctl config list"),
     (task_command.task_states_for_dag_run, "airflowctl tasks states-for-dag-run"),
+    (task_command.task_clear, "airflowctl tasks clear"),
 ]
 
 

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

@@ -99,6 +99,7 @@ def date_param():
     # Tasks commands
     'tasks states-for-dag-run example_bash_operator "manual__{date_param}"',
     'tasks states-for-dag-run example_bash_operator --logical-date "{date_param}"',
+    'tasks clear example_bash_operator --dag-run-id "manual__{date_param}" --task-ids runme_0 -o json',
     # Task Instances commands
     'taskinstances list example_bash_operator "manual__{date_param}"',
     # XCom commands - need a Dag run with completed tasks

airflow-ctl/docs/images/command_hashes.txt

@@ -1,4 +1,4 @@
-main:0460d9c03248bee26207b20b05aa36b9
+main:2c358f5fc894541cab11854dcd67658a
 assets:6419e20452692f577c4c6f570b74be0c
 auth:d79e9c7d00c432bdbcbc2a86e2e32053
 backfill:74c8737b0a62a86ed3605fa9e6165874
@@ -10,7 +10,7 @@ jobs:a5b644c5da8889443bb40ee10b599270
 pools:19efe105b9515ab1926ebcaf0e028d71
 providers:34502fe09dc0b8b0a13e7e46efdffda6
 taskinstances:7e323968c0b585287c2a4ab4339aee1f
-tasks:089b19625c893d189b5fc02b3abd547e
+tasks:e2ed0ab7f67ebacf4be87b5f396ff782
 variables:f8fc76d3d398b2780f4e97f7cd816646
 version:31f4efdf8de0dbaaa4fac71ff7efecc3
 plugins:4864fd8f356704bd2b3cd1aec3567e35

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

@@ -58,6 +58,7 @@
     ProvidersOperations,
     ServerResponseError,
     TaskInstancesOperations,
+    TasksOperations,
     VariablesOperations,
     VersionOperations,
     XComOperations,
@@ -457,6 +458,12 @@ def task_instances(self):
         """Operations related to task instances."""
         return TaskInstancesOperations(self)
 
+    @lru_cache()  # type: ignore[prop-decorator]
+    @property
+    def tasks(self):
+        """Operations related to tasks."""
+        return TasksOperations(self)
+
     @lru_cache()  # type: ignore[prop-decorator]
     @property
     def variables(self):

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

@@ -39,6 +39,7 @@
     BulkBodyPoolBody,
     BulkBodyVariableBody,
     BulkResponse,
+    ClearTaskInstancesBody,
     Config,
     ConnectionBody,
     ConnectionCollectionResponse,
@@ -787,6 +788,23 @@ def list(self, dag_id: str, dag_run_id: str) -> TaskInstanceCollectionResponse |
         )
 
 
+class TasksOperations(BaseOperations):
+    """Tasks operations."""
+
+    def clear(
+        self, dag_id: str, clear_task_instances: ClearTaskInstancesBody
+    ) -> TaskInstanceCollectionResponse | ServerResponseError:
+        """Clear task instances of a Dag; with dry_run (the default) only previews the affected task instances."""
+        try:
+            self.response = self.client.post(
+                f"dags/{dag_id}/clearTaskInstances",
+                json=clear_task_instances.model_dump(mode="json", exclude_none=True),
+            )
+            return TaskInstanceCollectionResponse.model_validate_json(self.response.content)
+        except ServerResponseError as e:
+            raise e
+
+
 class VariablesOperations(BaseOperations):
     """Variable operations."""
 

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

@@ -426,7 +426,21 @@ 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",
+        ]
+        # Datamodels whose generated bool flags follow the datamodel field defaults instead of
+        # defaulting to False, so the CLI keeps the API semantics (e.g. a bare ``tasks clear``
+        # must keep ``dry_run=True`` and preview instead of clearing).
+        self.field_bool_default_datamodels = ["ClearTaskInstancesBody"]
         self.exclude_operation_names = ["LoginOperations", "VersionOperations", "BaseOperations"]
         self.exclude_method_names = [
             "error",
@@ -593,6 +607,14 @@ def _create_positional_arg(
             help=arg_help,
         )
 
+    def _get_bool_arg_default(self, parameter_type: str, field_default: Any) -> bool | None:
+        """Get default for a generated bool flag: the datamodel field default for datamodels in ``field_bool_default_datamodels``, otherwise False."""
+        if parameter_type in self.field_bool_default_datamodels and (
+            field_default is None or isinstance(field_default, bool)
+        ):
+            return field_default
+        return False
+
     def _create_arg_for_non_primitive_type(
         self,
         parameter_type: str,
@@ -608,30 +630,24 @@ def _create_arg_for_non_primitive_type(
                 continue
             self.datamodels_extended_map[parameter_type].append(field)
             if type(field_type.annotation) is type:
-                commands.append(
-                    self._create_arg(
-                        arg_flags=("--" + self._sanitize_arg_parameter_key(field),),
-                        arg_type=self._python_type_from_string(field_type.annotation),
-                        arg_action=argparse.BooleanOptionalAction if field_type.annotation is bool else None,  # type: ignore
-                        arg_help=f"{field} for {parameter_key} operation",
-                        arg_default=False if field_type.annotation is bool else None,
-                    )
-                )
+                annotation = field_type.annotation
             else:
                 try:
                     annotation = field_type.annotation.__args__[0]
                 except AttributeError:
                     annotation = field_type.annotation
 
-                commands.append(
-                    self._create_arg(
-                        arg_flags=("--" + self._sanitize_arg_parameter_key(field),),
-                        arg_type=self._python_type_from_string(annotation),
-                        arg_action=argparse.BooleanOptionalAction if annotation is bool else None,  # type: ignore
-                        arg_help=f"{field} for {parameter_key} operation",
-                        arg_default=False if annotation is bool else None,
-                    )
+            commands.append(
+                self._create_arg(
+                    arg_flags=(f"--{self._sanitize_arg_parameter_key(field)}",),
+                    arg_type=self._python_type_from_string(annotation),
+                    arg_action=argparse.BooleanOptionalAction if annotation is bool else None,  # type: ignore
+                    arg_help=f"{field} for {parameter_key} operation",
+                    arg_default=self._get_bool_arg_default(parameter_type, field_type.default)
+                    if annotation is bool
+                    else None,
                 )
+            )
         return commands
 
     def _create_args_map_from_operation(self):
@@ -711,6 +727,20 @@ def _apply_datamodel_defaults(self, datamodel: type, params: dict) -> dict:
         ):
             params["logical_date"] = datetime.datetime.now(datetime.timezone.utc)
 
+        # Handle ClearTaskInstancesBody: --task-ids arrives as a single string but the API expects
+        # a list of task_id or [task_id, map_index]; accept comma-separated ids or a JSON list
+        if datamodel.__name__ == "ClearTaskInstancesBody" and isinstance(params.get("task_ids"), str):
+            raw_task_ids = params["task_ids"]
+            if raw_task_ids.lstrip().startswith("["):
+                try:
+                    params["task_ids"] = json.loads(raw_task_ids)
+                except json.JSONDecodeError as e:
+                    raise SystemExit(f"Invalid JSON list for --task-ids {raw_task_ids!r}: {e}")
+            else:
+                params["task_ids"] = [
+                    task_id.strip() for task_id in raw_task_ids.split(",") if task_id.strip()
+                ]
+
         return params
 
     def _create_func_map_from_operation(self):