REST catalog: implement lazy pagination generators for list_* methods

Replace the collect-then-return approach with proper generator functions
that yield results page by page. Extract per-page fetch logic into
dedicated helper methods (_fetch_tables_page, _fetch_views_page,
_fetch_namespaces_page) decorated with @Retry so authentication retries
work correctly per page.

Co-authored-by: Yuya Ebihara <ebyhry@gmail.com>

Author: 2 people

pyiceberg/catalog/rest/__init__.py

@@ -1038,6 +1038,14 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
         return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
 
     @retry(**_RETRY_ARGS)
+    def _fetch_tables_page(self, url: str, params: dict[str, str]) -> ListTablesResponse:
+        response = self._session.get(url, params=params)
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {404: NoSuchNamespaceError})
+        return ListTablesResponse.model_validate_json(response.text)
+
     @override
     def list_tables(self, namespace: str | Identifier) -> Iterator[Identifier]:
         self._check_endpoint(Capability.V1_LIST_TABLES)
@@ -1052,27 +1060,18 @@ def list_tables(self, namespace: str | Identifier) -> Iterator[Identifier]:
                 raise ValueError(f"{PAGE_SIZE} must be a positive integer")
             params["pageSize"] = str(page_size)
 
-        tables: list[Identifier] = []
         page_token: str | None = None
 
         while True:
             if page_token:
                 params["pageToken"] = page_token
-            response = self._session.get(url, params=params)
-            try:
-                response.raise_for_status()
-            except HTTPError as exc:
-                _handle_non_200_response(exc, {404: NoSuchNamespaceError})
-
-            parsed = ListTablesResponse.model_validate_json(response.text)
-            tables.extend([(*table.namespace, table.name) for table in parsed.identifiers])
+            parsed = self._fetch_tables_page(url, params)
+            yield from [(*table.namespace, table.name) for table in parsed.identifiers]
 
             if not parsed.next_page_token:
                 break
             page_token = parsed.next_page_token
 
-        return iter(tables)
-
     @retry(**_RETRY_ARGS)
     @override
     def load_table(self, identifier: str | Identifier) -> Table:
@@ -1151,10 +1150,18 @@ def _remove_catalog_name_from_table_request_identifier(self, table_request: Comm
         return table_request
 
     @retry(**_RETRY_ARGS)
+    def _fetch_views_page(self, url: str, params: dict[str, str]) -> ListViewsResponse:
+        response = self._session.get(url, params=params)
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {404: NoSuchNamespaceError})
+        return ListViewsResponse.model_validate_json(response.text)
+
     @override
     def list_views(self, namespace: str | Identifier) -> Iterator[Identifier]:
         if Capability.V1_LIST_VIEWS not in self._supported_endpoints:
-            return []
+            return
         namespace_tuple = self._check_valid_namespace_identifier(namespace)
         namespace_concat = self._encode_namespace_path(namespace_tuple)
         url = self.url(Endpoints.list_views, namespace=namespace_concat)
@@ -1166,28 +1173,18 @@ def list_views(self, namespace: str | Identifier) -> Iterator[Identifier]:
                 raise ValueError(f"{PAGE_SIZE} must be a positive integer")
             params["pageSize"] = str(page_size)
 
-        views: list[Identifier] = []
         page_token: str | None = None
 
         while True:
             if page_token:
                 params["pageToken"] = page_token
-
-            response = self._session.get(url, params=params)
-            try:
-                response.raise_for_status()
-            except HTTPError as exc:
-                _handle_non_200_response(exc, {404: NoSuchNamespaceError})
-
-            parsed = ListViewsResponse.model_validate_json(response.text)
-            views.extend([(*view.namespace, view.name) for view in parsed.identifiers])
+            parsed = self._fetch_views_page(url, params)
+            yield from [(*view.namespace, view.name) for view in parsed.identifiers]
 
             if not parsed.next_page_token:
                 break
             page_token = parsed.next_page_token
 
-        return iter(views)
-
     @retry(**_RETRY_ARGS)
     @override
     def load_view(self, identifier: str | Identifier) -> View:
@@ -1276,6 +1273,14 @@ def drop_namespace(self, namespace: str | Identifier) -> None:
             _handle_non_200_response(exc, {404: NoSuchNamespaceError, 409: NamespaceNotEmptyError})
 
     @retry(**_RETRY_ARGS)
+    def _fetch_namespaces_page(self, params: dict[str, str]) -> ListNamespaceResponse:
+        response = self._session.get(self.url(Endpoints.list_namespaces), params=params)
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {404: NoSuchNamespaceError})
+        return ListNamespaceResponse.model_validate_json(response.text)
+
     @override
     def list_namespaces(self, namespace: str | Identifier = ()) -> Iterator[Identifier]:
         self._check_endpoint(Capability.V1_LIST_NAMESPACES)
@@ -1288,30 +1293,20 @@ def list_namespaces(self, namespace: str | Identifier = ()) -> Iterator[Identifi
                 raise ValueError(f"{PAGE_SIZE} must be a positive integer")
             params["pageSize"] = str(page_size)
 
-        namespaces: list[Identifier] = []
         page_token: str | None = None
 
         while True:
             if namespace_tuple:
                 params["parent"] = self._encode_namespace_path(namespace_tuple)
             if page_token:
                 params["pageToken"] = page_token
-            response = self._session.get(self.url(Endpoints.list_namespaces), params=params)
-
-            try:
-                response.raise_for_status()
-            except HTTPError as exc:
-                _handle_non_200_response(exc, {404: NoSuchNamespaceError})
-
-            parsed = ListNamespaceResponse.model_validate_json(response.text)
-            namespaces.extend(parsed.namespaces)
+            parsed = self._fetch_namespaces_page(params)
+            yield from parsed.namespaces
 
             if not parsed.next_page_token:
                 break
             page_token = parsed.next_page_token
 
-        return iter(namespaces)
-
     @retry(**_RETRY_ARGS)
     @override
     def load_namespace_properties(self, namespace: str | Identifier) -> Properties: