From 08c484f152497f7efa29e7172ec99b85015bb67b Mon Sep 17 00:00:00 2001 From: itzzdev09 Date: Thu, 27 Aug 2026 22:21:06 +0530 Subject: [PATCH 1/4] Add state command to airflowctl tasks Adds `airflowctl tasks state`, the replacement for the deprecated `airflow tasks state` command, which is already marked `@deprecated_for_airflowctl("airflowctl tasks state")` but had no airflowctl counterpart. The command selects a Dag run with either `run_id` or `--logical-date` and prints the state of the task instance, reusing the existing `task_instances.get` endpoint. It follows the argument handling, error messages and map-index reporting already used by `tasks failed-deps`, and prints "None" for an unset state so scripts migrating off the deprecated command keep working. --- airflow-ctl/src/airflowctl/ctl/cli_config.py | 16 ++ .../airflowctl/ctl/commands/task_command.py | 31 +++ .../ctl/commands/test_task_command.py | 261 ++++++++++++++++++ 3 files changed, 308 insertions(+) diff --git a/airflow-ctl/src/airflowctl/ctl/cli_config.py b/airflow-ctl/src/airflowctl/ctl/cli_config.py index 8a2bfd2e15b4d..7b147b956ffd3 100755 --- a/airflow-ctl/src/airflowctl/ctl/cli_config.py +++ b/airflow-ctl/src/airflowctl/ctl/cli_config.py @@ -1199,6 +1199,22 @@ 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)." + ), + 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", diff --git a/airflow-ctl/src/airflowctl/ctl/commands/task_command.py b/airflow-ctl/src/airflowctl/ctl/commands/task_command.py index abd61b23dde99..001784acefb16 100644 --- a/airflow-ctl/src/airflowctl/ctl/commands/task_command.py +++ b/airflow-ctl/src/airflowctl/ctl/commands/task_command.py @@ -126,6 +126,37 @@ 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.""" + if (args.run_id is None) == (args.logical_date is None): + rich.print("[red]Provide either run_id or --logical-date, but not both[/red]") + sys.exit(1) + + run_id = args.run_id or _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.""" diff --git a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py index 2f6f23f6120ad..5a934904c7b8e 100644 --- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py +++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_task_command.py @@ -376,6 +376,267 @@ def test_failed_deps_reraises_non_404_dag_run_list_error(self): ) +class TestState: + parser = cli_parser.get_parser() + dag_id = "test_dag" + run_id = "test_run" + task_id = "test_task" + logical_date = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc) + + def _make_api_client(self, state: TaskInstanceState | None = None) -> mock.MagicMock: + api_client = mock.MagicMock() + api_client.dag_runs.list.return_value.dag_runs = [mock.MagicMock(dag_run_id=self.run_id)] + api_client.task_instances.get.return_value = TaskInstanceResponse( + id=uuid.uuid4(), + task_id=self.task_id, + dag_id=self.dag_id, + dag_run_id=self.run_id, + map_index=-1, + logical_date=self.logical_date, + run_after=self.logical_date, + state=state, + start_date=None, + end_date=None, + try_number=1, + max_tries=0, + task_display_name=self.task_id, + dag_display_name=self.dag_id, + pool="default_pool", + pool_slots=1, + executor_config="{}", + duration=None, + hostname=None, + unixname=None, + queue=None, + priority_weight=None, + operator=None, + operator_name=None, + queued_when=None, + scheduled_when=None, + pid=None, + executor=None, + note=None, + rendered_map_index=None, + trigger=None, + triggerer_job=None, + dag_version=None, + ) + return api_client + + @pytest.mark.parametrize( + "task_instance_state", + [TaskInstanceState.SUCCESS, TaskInstanceState.FAILED, TaskInstanceState.RUNNING], + ) + def test_state_by_run_id(self, task_instance_state, capsys): + api_client = self._make_api_client(state=task_instance_state) + + task_command.state( + self.parser.parse_args(["tasks", "state", self.dag_id, self.task_id, self.run_id]), + api_client=api_client, + ) + + api_client.task_instances.get.assert_called_once_with( + dag_id=self.dag_id, + dag_run_id=self.run_id, + task_id=self.task_id, + map_index=-1, + suppress_error_log=True, + ) + assert capsys.readouterr().out == f"{task_instance_state.value}\n" + + def test_state_without_state_prints_none(self, capsys): + api_client = self._make_api_client(state=None) + + task_command.state( + self.parser.parse_args(["tasks", "state", self.dag_id, self.task_id, self.run_id]), + api_client=api_client, + ) + + assert capsys.readouterr().out == "None\n" + + def test_state_with_map_index(self, capsys): + api_client = self._make_api_client(state=TaskInstanceState.SUCCESS) + + task_command.state( + self.parser.parse_args( + ["tasks", "state", self.dag_id, self.task_id, self.run_id, "--map-index", "3"] + ), + api_client=api_client, + ) + + api_client.task_instances.get.assert_called_once_with( + dag_id=self.dag_id, + dag_run_id=self.run_id, + task_id=self.task_id, + map_index=3, + suppress_error_log=True, + ) + assert capsys.readouterr().out == "success\n" + + def test_state_by_logical_date(self, capsys): + api_client = self._make_api_client(state=TaskInstanceState.SUCCESS) + + task_command.state( + self.parser.parse_args( + [ + "tasks", + "state", + self.dag_id, + self.task_id, + "--logical-date", + self.logical_date.isoformat(), + ] + ), + api_client=api_client, + ) + + api_client.dag_runs.list.assert_called_once_with( + dag_id=self.dag_id, + logical_date_gte=self.logical_date, + logical_date_lte=self.logical_date, + order_by="-id", + limit=1, + suppress_error_log=True, + ) + api_client.task_instances.get.assert_called_once_with( + dag_id=self.dag_id, + dag_run_id=self.run_id, + task_id=self.task_id, + map_index=-1, + suppress_error_log=True, + ) + assert capsys.readouterr().out == "success\n" + + @pytest.mark.parametrize( + "extra_args", + [ + [], + ["test_run", "--logical-date", "2025-01-01T00:00:00+00:00"], + ], + ids=["neither", "both"], + ) + def test_state_requires_exactly_one_of_run_id_and_logical_date(self, extra_args, capsys): + api_client = self._make_api_client() + + with pytest.raises(SystemExit, match="1"): + task_command.state( + self.parser.parse_args(["tasks", "state", self.dag_id, self.task_id, *extra_args]), + api_client=api_client, + ) + + api_client.task_instances.get.assert_not_called() + assert _normalize_rich_output(capsys.readouterr().out) == ( + "Provide either run_id or --logical-date, but not both" + ) + + @pytest.mark.parametrize( + ("logical_date", "expected_message"), + [ + ("not-a-date", "Invalid --logical-date: 'not-a-date'"), + ("2025-01-01T00:00:00", "--logical-date must include a timezone offset"), + ], + ids=["unparsable", "naive"], + ) + def test_state_rejects_bad_logical_date(self, logical_date, expected_message, capsys): + api_client = self._make_api_client() + + with pytest.raises(SystemExit, match="1"): + task_command.state( + self.parser.parse_args( + ["tasks", "state", self.dag_id, self.task_id, "--logical-date", logical_date] + ), + api_client=api_client, + ) + + api_client.dag_runs.list.assert_not_called() + assert _normalize_rich_output(capsys.readouterr().out) == expected_message + + @pytest.mark.parametrize( + ("extra_args", "expected_message"), + [ + ( + [], + "Task instance for task 'test_task' in Dag run 'test_run' of Dag 'test_dag' not found", + ), + ( + ["--map-index", "3"], + "Task instance for task 'test_task' with map index 3 in Dag run 'test_run' " + "of Dag 'test_dag' not found", + ), + ], + ) + def test_state_task_instance_not_found(self, extra_args, expected_message, capsys): + api_client = self._make_api_client() + api_client.task_instances.get.side_effect = _make_server_error(404) + + with pytest.raises(SystemExit, match="1"): + task_command.state( + self.parser.parse_args( + ["tasks", "state", self.dag_id, self.task_id, self.run_id, *extra_args] + ), + api_client=api_client, + ) + + assert _normalize_rich_output(capsys.readouterr().out) == expected_message + + @pytest.mark.parametrize("list_failure", ["no_matching_run", "dag_not_found_404"]) + def test_state_dag_run_not_found_by_logical_date(self, list_failure, capsys): + api_client = self._make_api_client() + if list_failure == "no_matching_run": + api_client.dag_runs.list.return_value.dag_runs = [] + else: + api_client.dag_runs.list.side_effect = _make_server_error(404) + + with pytest.raises(SystemExit, match="1"): + task_command.state( + self.parser.parse_args( + [ + "tasks", + "state", + self.dag_id, + self.task_id, + "--logical-date", + self.logical_date.isoformat(), + ] + ), + api_client=api_client, + ) + + api_client.task_instances.get.assert_not_called() + assert _normalize_rich_output(capsys.readouterr().out) == ( + "Dag run for test_dag with logical date '2025-01-01T00:00:00+00:00' not found" + ) + + def test_state_reraises_non_404_error(self): + api_client = self._make_api_client() + api_client.task_instances.get.side_effect = _make_server_error(500) + + with pytest.raises(ServerResponseError): + task_command.state( + self.parser.parse_args(["tasks", "state", self.dag_id, self.task_id, self.run_id]), + api_client=api_client, + ) + + def test_state_reraises_non_404_dag_run_list_error(self): + api_client = self._make_api_client() + api_client.dag_runs.list.side_effect = _make_server_error(500) + + with pytest.raises(ServerResponseError): + task_command.state( + self.parser.parse_args( + [ + "tasks", + "state", + self.dag_id, + self.task_id, + "--logical-date", + self.logical_date.isoformat(), + ] + ), + api_client=api_client, + ) + + class TestStatesForDagRun: parser = cli_parser.get_parser() dag_id = "test_dag" From 3e2b91220e797a9557ec9f1ff5a001c9c302b59c Mon Sep 17 00:00:00 2001 From: itzzdev09 Date: Fri, 28 Aug 2026 03:10:36 +0530 Subject: [PATCH 2/4] Address review: explicit run selection, clarify help, add newsfragment - Make the run_id / --logical-date validation explicit with named booleans and an if/else instead of the compact 'or' shortcut. - Note in the command description that --logical-date matches the Dag run with exactly that logical date. - Add the newsfragment for the new command. Co-authored-by: SameerMesiah97 <75502260+SameerMesiah97@users.noreply.github.com> --- airflow-core/newsfragments/72163.feature.rst | 1 + airflow-ctl/src/airflowctl/ctl/cli_config.py | 3 ++- .../src/airflowctl/ctl/commands/task_command.py | 10 ++++++++-- 3 files changed, 11 insertions(+), 3 deletions(-) create mode 100644 airflow-core/newsfragments/72163.feature.rst diff --git a/airflow-core/newsfragments/72163.feature.rst b/airflow-core/newsfragments/72163.feature.rst new file mode 100644 index 0000000000000..dd69f42813319 --- /dev/null +++ b/airflow-core/newsfragments/72163.feature.rst @@ -0,0 +1 @@ +Add ``airflowctl tasks state`` command to get the state of a task instance, selected by ``run_id`` or ``--logical-date``. diff --git a/airflow-ctl/src/airflowctl/ctl/cli_config.py b/airflow-ctl/src/airflowctl/ctl/cli_config.py index 7b147b956ffd3..a0b6cfa176542 100755 --- a/airflow-ctl/src/airflowctl/ctl/cli_config.py +++ b/airflow-ctl/src/airflowctl/ctl/cli_config.py @@ -1204,7 +1204,8 @@ def merge_commands( 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)." + "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=( diff --git a/airflow-ctl/src/airflowctl/ctl/commands/task_command.py b/airflow-ctl/src/airflowctl/ctl/commands/task_command.py index 001784acefb16..6732e5db25327 100644 --- a/airflow-ctl/src/airflowctl/ctl/commands/task_command.py +++ b/airflow-ctl/src/airflowctl/ctl/commands/task_command.py @@ -129,11 +129,17 @@ 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.""" - if (args.run_id is None) == (args.logical_date is None): + 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) - run_id = args.run_id or _find_run_id_by_logical_date(api_client, args.dag_id, args.logical_date) + 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( From 9b6b10c1082fd58f0f7b4bfb7637423b40a58622 Mon Sep 17 00:00:00 2001 From: itzzdev09 Date: Fri, 28 Aug 2026 03:51:28 +0530 Subject: [PATCH 3/4] Regenerate airflowctl tasks help image The generate-airflowctl-help-images prek hook regenerates the CLI help SVGs from cli_config.py, so adding the 'tasks state' command left the checked-in output_tasks.svg and its hash stale and failed static checks. --- airflow-ctl/docs/images/command_hashes.txt | 2 +- airflow-ctl/docs/images/output_tasks.svg | 78 ++++++++++++---------- 2 files changed, 42 insertions(+), 38 deletions(-) diff --git a/airflow-ctl/docs/images/command_hashes.txt b/airflow-ctl/docs/images/command_hashes.txt index 4e48ad37fee6f..b226e7437654c 100644 --- a/airflow-ctl/docs/images/command_hashes.txt +++ b/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 diff --git a/airflow-ctl/docs/images/output_tasks.svg b/airflow-ctl/docs/images/output_tasks.svg index 9f30d658f385e..08d22e642ed12 100644 --- a/airflow-ctl/docs/images/output_tasks.svg +++ b/airflow-ctl/docs/images/output_tasks.svg @@ -1,4 +1,4 @@ - + - - + + - + - + - + - + - + - + - + - + - + - + - + - + + + + - + - + - - Usage:airflowctl tasks [-h] COMMAND... - -Perform Tasks operations - -Positional Arguments: -COMMAND -clearClear task instances of a Dag by its ID -failed-depsReturns the unmet dependencies for a task instance -states-for-dag-run -Get the status of all task instances in a Dag run - -Options: --h, --helpshow this help message and exit + + Usage:airflowctl tasks [-h] COMMAND... + +Perform Tasks operations + +Positional Arguments: +COMMAND +clearClear task instances of a Dag by its ID +failed-depsReturns the unmet dependencies for a task instance +stateGet the state of a task instance +states-for-dag-run +Get the status of all task instances in a Dag run + +Options: +-h, --helpshow this help message and exit From 5f7232f41c9e03090f691162dfda37fa3d97fbc1 Mon Sep 17 00:00:00 2001 From: itzzdev09 Date: Sun, 30 Aug 2026 09:28:50 +0530 Subject: [PATCH 4/4] Remove newsfragment: airflowctl is its own package, not core Co-Authored-By: Claude Sonnet 5 --- airflow-core/newsfragments/72163.feature.rst | 1 - 1 file changed, 1 deletion(-) delete mode 100644 airflow-core/newsfragments/72163.feature.rst diff --git a/airflow-core/newsfragments/72163.feature.rst b/airflow-core/newsfragments/72163.feature.rst deleted file mode 100644 index dd69f42813319..0000000000000 --- a/airflow-core/newsfragments/72163.feature.rst +++ /dev/null @@ -1 +0,0 @@ -Add ``airflowctl tasks state`` command to get the state of a task instance, selected by ``run_id`` or ``--logical-date``.