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

Author: ColtenOuO

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',

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

airflow-ctl/docs/images/output_tasks.svg

@@ -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."""

airflow-ctl/src/airflowctl/api/operations.py

@@ -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"

airflow-ctl/src/airflowctl/ctl/help_texts.yaml

@@ -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"