Add airflowctl tasks state command

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

@@ -107,6 +107,8 @@ def date_param():
     'tasks failed-deps example_bash_operator runme_0 --logical-date "{date_param}"',
     'tasks states-for-dag-run example_bash_operator "manual__{date_param}"',
     'tasks states-for-dag-run example_bash_operator --logical-date "{date_param}"',
+    'tasks state example_bash_operator runme_0 "manual__{date_param}"',
+    'tasks state example_bash_operator runme_0 --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 get example_bash_operator "manual__{date_param}" runme_0',

airflow-ctl/docs/images/command_hashes.txt

@@ -10,7 +10,7 @@ jobs:d4af478f28dae48ee18d43b998edc345
 pools:19efe105b9515ab1926ebcaf0e028d71
 providers:34502fe09dc0b8b0a13e7e46efdffda6
 taskinstances:bea84117114c2438eb7e7026f6bb7042
-tasks:ea587dc805cadbce81cd640d357f2661
+tasks:eb70701dfe1b9baeda39d5eea07fde5f
 variables:f8fc76d3d398b2780f4e97f7cd816646
 version:31f4efdf8de0dbaaa4fac71ff7efecc3
 plugins:4864fd8f356704bd2b3cd1aec3567e35

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

@@ -1228,6 +1228,23 @@ def merge_commands(
             ARG_MAP_INDEX,
         ),
     ),
+    ActionCommand(
+        name="state",
+        help="Get the state of a task instance",
+        description=(
+            "Get the state of a task instance. "
+            "Select the run with either run_id or --logical-date (pass exactly one). "
+            "Prints the state value, or None when the task instance has no state yet."
+        ),
+        func=lazy_load_command("airflowctl.ctl.commands.task_command.state"),
+        args=(
+            ARG_DAG_ID,
+            ARG_TASK_ID,
+            ARG_RUN_ID,
+            ARG_LOGICAL_DATE,
+            ARG_MAP_INDEX,
+        ),
+    ),
     ActionCommand(
         name="states-for-dag-run",
         help="Get the status of all task instances in a Dag run",

airflow-ctl/src/airflowctl/ctl/commands/task_command.py

@@ -31,6 +31,29 @@
     from airflowctl.api.datamodels.generated import TaskInstanceResponse
 
 
+def _is_mapped_task_error(error: ServerResponseError) -> bool:
+    """Whether the API rejected the lookup because the task is mapped and no map index was given."""
+    try:
+        detail = error.response.json().get("detail")
+    except ValueError:
+        return False
+    return isinstance(detail, str) and "is mapped" in detail
+
+
+def _task_instance_not_found_message(
+    dag_id: str, run_id: str, task_id: str, map_index: int, error: ServerResponseError
+) -> str:
+    """Build the message shown when a task instance is not found."""
+    map_index_part = f" with map index {map_index}" if map_index >= 0 else ""
+    message = (
+        f"Task instance for task {task_id!r}{map_index_part} in Dag run "
+        f"{run_id!r} of Dag {dag_id!r} not found"
+    )
+    if map_index < 0 and _is_mapped_task_error(error):
+        message += ". The task is mapped; pass --map-index to select one of its task instances"
+    return message
+
+
 def _format_task_instance(ti: TaskInstanceResponse, has_mapped_instances: bool) -> dict[str, str]:
     data = {
         "dag_id": ti.dag_id,
@@ -75,10 +98,8 @@ def failed_deps(args, api_client=NEW_API_CLIENT) -> None:
         )
     except ServerResponseError as e:
         if e.response.status_code == 404:
-            map_index_part = f" with map index {args.map_index}" if args.map_index >= 0 else ""
             rich.print(
-                f"[red]Task instance for task {args.task_id!r}{map_index_part} in Dag run "
-                f"{run_id!r} of Dag {args.dag_id!r} not found[/red]"
+                f"[red]{_task_instance_not_found_message(args.dag_id, run_id, args.task_id, args.map_index, e)}[/red]"
             )
             sys.exit(1)
         raise
@@ -111,3 +132,29 @@ def states_for_dag_run(args, api_client=NEW_API_CLIENT) -> None:
         data=[_format_task_instance(ti, has_mapped_instances) for ti in task_instances],
         output=args.output,
     )
+
+
+@provide_api_client(kind=ClientKind.CLI)
+def state(args, api_client=NEW_API_CLIENT) -> None:
+    """Get the state of a task instance."""
+    run_id = resolve_dag_run_id(api_client, args)
+
+    try:
+        task_instance = api_client.task_instances.get(
+            dag_id=args.dag_id,
+            dag_run_id=run_id,
+            task_id=args.task_id,
+            map_index=args.map_index,
+            suppress_error_log=True,
+        )
+    except ServerResponseError as e:
+        if e.response.status_code == 404:
+            rich.print(
+                f"[red]{_task_instance_not_found_message(args.dag_id, run_id, args.task_id, args.map_index, e)}[/red]"
+            )
+            sys.exit(1)
+        raise
+
+    # Unset states print as None to stay drop-in compatible with the deprecated
+    # `airflow tasks state`, which prints its nullable `ti.state` column directly.
+    print(task_instance.state.value if task_instance.state else None)