From e26b381a10e8f91f9231f03c7dbf38e903b1cfb2 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 7e5c94014f316..4b2217355d9fb 100644 --- a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py +++ b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py @@ -108,6 +108,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 bd58cb3c8cd4f..c874ad5503f86 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: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 c5f0dc392f02c..d13bf7cc42a28 100644 --- a/airflow-ctl/src/airflowctl/api/operations.py +++ b/airflow-ctl/src/airflowctl/api/operations.py @@ -72,6 +72,7 @@ ProviderCollectionResponse, QueuedEventCollectionResponse, QueuedEventResponse, + TaskCollectionResponse, TaskDependencyCollectionResponse, TaskInstanceCollectionResponse, TaskInstanceResponse, @@ -772,6 +773,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 b2e5a4129cc53..f755ebf8ae258 100644 --- a/airflow-ctl/src/airflowctl/ctl/help_texts.yaml +++ b/airflow-ctl/src/airflowctl/ctl/help_texts.yaml @@ -98,6 +98,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 f1eaf2e89efe1..9f555758903c2 100644 --- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py +++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py @@ -95,11 +95,13 @@ QueuedEventCollectionResponse, QueuedEventResponse, ReprocessBehavior, + TaskCollectionResponse, TaskDependencyCollectionResponse, TaskDependencyResponse, TaskInstanceCollectionResponse, TaskInstanceResponse, TaskInstanceState, + TaskResponse, TriggerDAGRunPostBody, VariableBody, VariableCollectionResponse, @@ -1998,6 +2000,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) @@ -2014,6 +2050,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 45d55f0803434..55098ad4ed98d 100644 --- a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py +++ b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py @@ -860,6 +860,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"), [ @@ -895,6 +910,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):