Fix airflowctl dagrun list crash when --state is omitted

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

@@ -606,8 +606,8 @@ def get(self, dag_id: str, dag_run_id: str) -> DAGRunResponse | ServerResponseEr
 
     def list(
         self,
-        state: str,
-        limit: int,
+        state: str | None = None,
+        limit: int = 100,
         start_date: datetime.datetime | None = None,
         end_date: datetime.datetime | None = None,
         dag_id: str | None = None,
@@ -616,7 +616,7 @@ def list(
         List dag runs (at most `limit` results).
 
         Args:
-            state: Filter dag runs by state
+            state: Filter dag runs by state (optional; no filter applied when omitted)
             start_date: Filter dag runs by start date (optional)
             end_date: Filter dag runs by end date (optional)
             limit: Limit the number of results returned
@@ -626,10 +626,9 @@ def list(
         if not dag_id:
             dag_id = "~"
 
-        params: dict[str, Any] = {
-            "state": str(state),
-            "limit": limit,
-        }
+        params: dict[str, Any] = {"limit": limit}
+        if state is not None:
+            params["state"] = str(state)
         if start_date is not None:
             params["start_date"] = start_date.isoformat()
         if end_date is not None:

airflow-ctl/tests/airflow_ctl/api/test_operations.py

@@ -1139,6 +1139,20 @@ def handle_request(request: httpx.Request) -> httpx.Response:
         )
         assert response == self.dag_run_collection_response
 
+    def test_list_without_state_does_not_send_state_param(self):
+        """`state` is optional: omitting it must not send ``state=None`` to the API."""
+        captured_params: dict[str, str] = {}
+
+        def handle_request(request: httpx.Request) -> httpx.Response:
+            captured_params.update(dict(request.url.params))
+            return httpx.Response(200, json=json.loads(self.dag_run_collection_response.model_dump_json()))
+
+        client = make_api_client(transport=httpx.MockTransport(handle_request))
+        response = client.dag_runs.list(limit=5)
+        assert response == self.dag_run_collection_response
+        assert "state" not in captured_params
+        assert captured_params["limit"] == "5"
+
 
 class TestJobsOperations:
     job_response = JobResponse(