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):