mirror of
https://github.com/apache/airflow.git
synced 2026-11-04 07:12:48 +00:00
Fix airflowctl list pagination when the server caps the page size (#76268)
This commit is contained in:
1 parent
2ab01dae67
commit
5308480f2e
2 files changed
+54
-6
No files matched your search
@@ -332,20 +332,26 @@ class BaseOperations:
|
||||
total_entries = first_pass.total_entries
|
||||
# An older server that omits total_entries leaves it None; treat that as "no more
|
||||
# pages" rather than raising on the comparison below.
|
||||
if total_entries is None or total_entries < limit:
|
||||
if total_entries is None:
|
||||
return first_pass
|
||||
found_key = None
|
||||
for key, value in first_pass.model_dump().items():
|
||||
if key != "total_entries" and isinstance(value, list):
|
||||
found_key = key
|
||||
break
|
||||
if found_key is None:
|
||||
return first_pass
|
||||
entry_list = getattr(first_pass, found_key)
|
||||
offset = offset + limit
|
||||
while offset < total_entries:
|
||||
# The server may cap ``limit`` (``api.maximum_page_limit``), so advance by the number of
|
||||
# entries received rather than by the requested limit, or the capped-off entries are skipped.
|
||||
page_size = len(entry_list)
|
||||
offset = offset + page_size
|
||||
while page_size and offset < total_entries:
|
||||
self.response = self.client.get(path, params={**shared_params, "offset": offset})
|
||||
entry = validate_response(self.response.content, data_model)
|
||||
offset = offset + limit
|
||||
entry_list.extend(getattr(entry, found_key))
|
||||
page_entries = getattr(validate_response(self.response.content, data_model), found_key)
|
||||
entry_list.extend(page_entries)
|
||||
page_size = len(page_entries)
|
||||
offset = offset + page_size
|
||||
return data_model(**{found_key: entry_list, "total_entries": total_entries})
|
||||
|
||||
|
||||
|
||||
@@ -305,6 +305,48 @@ class TestBaseOperations:
|
||||
for call in mock_client.get.call_args_list:
|
||||
assert call.kwargs["params"]["limit"] == 2
|
||||
|
||||
def test_execute_list_advances_by_entries_received_when_server_caps_limit(self):
|
||||
"""A server with a lower ``api.maximum_page_limit`` returns fewer entries than requested;
|
||||
paging must continue from what was received or the entries in between are skipped."""
|
||||
mock_client = Mock(spec=Client)
|
||||
mock_client.get.side_effect = [
|
||||
Mock(content=json.dumps({"hellos": [{"name": "a"}, {"name": "b"}], "total_entries": 5})),
|
||||
Mock(content=json.dumps({"hellos": [{"name": "c"}, {"name": "d"}], "total_entries": 5})),
|
||||
Mock(content=json.dumps({"hellos": [{"name": "e"}], "total_entries": 5})),
|
||||
]
|
||||
base_operation = BaseOperations(client=mock_client)
|
||||
|
||||
response = base_operation.execute_list(path="hello", data_model=HelloCollectionResponse, limit=50)
|
||||
|
||||
assert [hello.name for hello in response.hellos] == ["a", "b", "c", "d", "e"]
|
||||
assert [call.kwargs["params"]["offset"] for call in mock_client.get.call_args_list] == [0, 2, 4]
|
||||
|
||||
def test_execute_list_stops_when_server_returns_an_empty_page(self):
|
||||
mock_client = Mock()
|
||||
mock_client.get.side_effect = [
|
||||
Mock(content=json.dumps({"hellos": [{"name": "a"}], "total_entries": 3})),
|
||||
Mock(content=json.dumps({"hellos": [], "total_entries": 3})),
|
||||
]
|
||||
base_operation = BaseOperations(client=mock_client)
|
||||
|
||||
response = base_operation.execute_list(path="hello", data_model=HelloCollectionResponse, limit=50)
|
||||
|
||||
assert [hello.name for hello in response.hellos] == ["a"]
|
||||
assert mock_client.get.call_count == 2
|
||||
|
||||
def test_execute_list_returns_first_page_when_response_has_no_entries_list(self):
|
||||
class _CountOnlyResponse(BaseModel):
|
||||
total_entries: int
|
||||
|
||||
mock_client = Mock()
|
||||
mock_client.get.return_value = Mock(content=json.dumps({"total_entries": 5}))
|
||||
base_operation = BaseOperations(client=mock_client)
|
||||
|
||||
response = base_operation.execute_list(path="hello", data_model=_CountOnlyResponse, limit=2)
|
||||
|
||||
assert response == _CountOnlyResponse(total_entries=5)
|
||||
assert mock_client.get.call_count == 1
|
||||
|
||||
def test_execute_list_sends_offset_to_server(self):
|
||||
"""``offset`` must reach the server on the first request, not only on subsequent pages."""
|
||||
rows = [{"name": name} for name in "abcdef"]
|
||||
|
||||
Reference in new issue
Block a user