From 5424ba75c1c27f636e09b161d5bd87e46bcf1176 Mon Sep 17 00:00:00 2001 From: ColtenOuO Date: Sun, 26 Jul 2026 12:54:58 +0000 Subject: [PATCH] Add airflowctl tasks list command Listing a Dag's tasks currently requires the local `airflow tasks list` command, which parses Dag files on the caller's machine and needs direct metadata database access. Exposing it through airflowctl lets operators list tasks against the Public API under RBAC instead, per AIP-94. related: #68402 --- .../test_airflowctl_commands.py | 2 + airflow-ctl/docs/images/command_hashes.txt | 2 +- airflow-ctl/docs/images/output_tasks.svg | 78 ++++++++++--------- airflow-ctl/src/airflowctl/api/operations.py | 6 ++ .../src/airflowctl/ctl/help_texts.yaml | 1 + .../tests/airflow_ctl/api/test_operations.py | 53 +++++++++++++ .../tests/airflow_ctl/ctl/test_cli_config.py | 16 ++++ 7 files changed, 120 insertions(+), 38 deletions(-) diff --git a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py index 14a500c4d5bf7..7137c6f5f70db 100644 --- a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py +++ b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py @@ -102,6 +102,8 @@ def 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 clear example_bash_operator --dag-run-id "manual__{date_param}" --task-ids runme_0 -o json', + "tasks list example_bash_operator", + "tasks list example_bash_operator --order-by=-task_id", # Task Instances commands 'taskinstances get example_bash_operator "manual__{date_param}" runme_0', 'taskinstances get-dependencies example_bash_operator "manual__{date_param}" runme_0', diff --git a/airflow-ctl/docs/images/command_hashes.txt b/airflow-ctl/docs/images/command_hashes.txt index 65c22154bc3c0..fd93d19791d54 100644 --- a/airflow-ctl/docs/images/command_hashes.txt +++ b/airflow-ctl/docs/images/command_hashes.txt @@ -10,7 +10,7 @@ jobs:a5b644c5da8889443bb40ee10b599270 pools:19efe105b9515ab1926ebcaf0e028d71 providers:34502fe09dc0b8b0a13e7e46efdffda6 taskinstances:bea84117114c2438eb7e7026f6bb7042 -tasks:ea587dc805cadbce81cd640d357f2661 +tasks:78f8bf0f0fbad664c37aa7489aab478f 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..16fd3b41f0866 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 +listList all tasks of a Dag +states-for-dag-run +Get the status of all task instances in a Dag run + +Options: +-h, --helpshow this help message and exit diff --git a/airflow-ctl/src/airflowctl/api/operations.py b/airflow-ctl/src/airflowctl/api/operations.py index ffd9b766984a1..71364c5b751a5 100644 --- a/airflow-ctl/src/airflowctl/api/operations.py +++ b/airflow-ctl/src/airflowctl/api/operations.py @@ -69,6 +69,7 @@ ProviderCollectionResponse, QueuedEventCollectionResponse, QueuedEventResponse, + TaskCollectionResponse, TaskDependencyCollectionResponse, TaskInstanceCollectionResponse, TaskInstanceResponse, @@ -736,6 +737,11 @@ def clear( ) return TaskInstanceCollectionResponse.model_validate_json(self.response.content) + def list(self, dag_id: str, order_by: str | None = None) -> TaskCollectionResponse | ServerResponseError: + """List tasks of a Dag.""" + self.response = self.client.get(f"dags/{dag_id}/tasks", params=_build_query_params(order_by=order_by)) + return TaskCollectionResponse.model_validate_json(self.response.content) + class VariablesOperations(BaseOperations): """Variable operations.""" diff --git a/airflow-ctl/src/airflowctl/ctl/help_texts.yaml b/airflow-ctl/src/airflowctl/ctl/help_texts.yaml index 08626e0093096..5686e9b66ca08 100644 --- a/airflow-ctl/src/airflowctl/ctl/help_texts.yaml +++ b/airflow-ctl/src/airflowctl/ctl/help_texts.yaml @@ -93,6 +93,7 @@ taskinstances: tasks: clear: "Clear task instances of a Dag by its ID" + list: "List all tasks of a Dag" variables: get: "Retrieve a variable by its key" diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py b/airflow-ctl/tests/airflow_ctl/api/test_operations.py index 4868841bc2fd2..efdcb809e8e4e 100644 --- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py +++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py @@ -92,11 +92,13 @@ QueuedEventCollectionResponse, QueuedEventResponse, ReprocessBehavior, + TaskCollectionResponse, TaskDependencyCollectionResponse, TaskDependencyResponse, TaskInstanceCollectionResponse, TaskInstanceResponse, TaskInstanceState, + TaskResponse, TriggerDAGRunPostBody, VariableBody, VariableCollectionResponse, @@ -1829,6 +1831,40 @@ class TestTasksOperations: task_instances=[task_instance_response], total_entries=1, ) + task_collection_response = TaskCollectionResponse( + tasks=[ + TaskResponse( + task_id="task_1", + task_display_name="task_1", + owner="airflow", + start_date=None, + end_date=None, + trigger_rule=None, + depends_on_past=False, + wait_for_downstream=False, + retries=None, + queue=None, + pool=None, + pool_slots=None, + execution_timeout=None, + retry_delay=None, + retry_exponential_backoff=0, + priority_weight=None, + weight_rule=None, + ui_color=None, + ui_fgcolor=None, + template_fields=None, + downstream_task_ids=None, + doc_md=None, + operator_name=None, + params=None, + class_ref=None, + is_mapped=None, + extra_links=[], + ) + ], + total_entries=1, + ) def test_clear(self): expected_body = self.clear_task_instances.model_dump(mode="json", exclude_none=True) @@ -1845,6 +1881,23 @@ def handle_request(request: httpx.Request) -> httpx.Response: response = client.tasks.clear(self.dag_id, self.clear_task_instances) assert response == self.task_instance_collection_response + @pytest.mark.parametrize( + ("order_by", "expected_params"), + [ + pytest.param(None, {}, id="no-order-by"), + pytest.param("-task_id", {"order_by": "-task_id"}, id="with-order-by"), + ], + ) + def test_list(self, order_by, expected_params): + def handle_request(request: httpx.Request) -> httpx.Response: + assert request.url.path == f"/api/v2/dags/{self.dag_id}/tasks" + assert dict(request.url.params) == expected_params + return httpx.Response(200, json=json.loads(self.task_collection_response.model_dump_json())) + + client = make_api_client(transport=httpx.MockTransport(handle_request)) + response = client.tasks.list(self.dag_id, order_by=order_by) + assert response == self.task_collection_response + class TestVariablesOperations: key = "key" diff --git a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py index fcbd3d748b2f2..e75afb1f4cd8b 100644 --- a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py +++ b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py @@ -801,6 +801,21 @@ def test_tasks_clear_args_follow_datamodel_defaults(self): assert args_by_flag["--task-ids"].kwargs["type"] is str assert "--output" in args_by_flag + def test_tasks_list_args(self): + command_factory = CommandFactory() + tasks_group = next( + group_command for group_command in command_factory.group_commands if group_command.name == "tasks" + ) + list_command = next( + sub_command for sub_command in tasks_group.subcommands if sub_command.name == "list" + ) + args_by_flag = {arg.flags[0]: arg for arg in list_command.args} + + assert "dag_id" in args_by_flag, "required path parameter should be positional" + assert args_by_flag["--order-by"].kwargs["type"] is str + assert args_by_flag["--order-by"].kwargs["default"] is None + assert "--output" in args_by_flag + @pytest.mark.parametrize( ("raw_task_ids", "expected_task_ids"), [ @@ -836,6 +851,7 @@ def test_apply_datamodel_defaults_clear_task_instances_invalid_json_task_ids(sel ("taskinstances", "get", "Get a task instance for a given Dag run"), ("taskinstances", "get-dependencies", "Get unmet scheduler dependencies for a task instance"), ("tasks", "clear", "Clear task instances of a Dag by its ID"), + ("tasks", "list", "List all tasks of a Dag"), ], ) def test_help_texts_used_for_auto_generated_commands(self, group_name, subcommand_name, expected_help):