Add state command to airflowctl tasks

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

@@ -1199,6 +1199,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). "
+            "--logical-date matches the Dag run with exactly that logical date."
+        ),
+        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

@@ -126,6 +126,43 @@ def failed_deps(args, api_client=NEW_API_CLIENT) -> None:
         )
 
 
+@provide_api_client(kind=ClientKind.CLI)
+def state(args, api_client=NEW_API_CLIENT) -> None:
+    """Get the state of a task instance."""
+    has_run_id = args.run_id is not None
+    has_logical_date = args.logical_date is not None
+    # Rejects both "neither given" and "both given".
+    if has_run_id == has_logical_date:
+        rich.print("[red]Provide either run_id or --logical-date, but not both[/red]")
+        sys.exit(1)
+
+    if has_run_id:
+        run_id = args.run_id
+    else:
+        run_id = _find_run_id_by_logical_date(api_client, args.dag_id, args.logical_date)
+
+    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:
+            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]"
+            )
+            sys.exit(1)
+        raise
+
+    # "None" keeps the output of the deprecated ``airflow tasks state`` for an unset state.
+    print(task_instance.state.value if task_instance.state else "None")
+
+
 @provide_api_client(kind=ClientKind.CLI)
 def states_for_dag_run(args, api_client=NEW_API_CLIENT) -> None:
     """Get the status of all task instances in a Dag run."""