REST: Add pagination support for list_namespaces (#3347)

<!--
Thanks for opening a pull request!
-->

<!-- In the case this PR will resolve an issue, please replace
${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
<!-- Closes #${GITHUB_ISSUE_ID} -->

# Rationale for this change

Follows REST catalog spec: 

- /v1/{prefix}/namespaces:
https://github.com/apache/iceberg/blob/e7a5a87f26f9de5b200254155aa037368b13a29c/open-api/rest-catalog-open-api.yaml#L250-L306
- ListNamespacesResponse:
https://github.com/apache/iceberg/blob/e7a5a87f26f9de5b200254155aa037368b13a29c/open-api/rest-catalog-open-api.yaml#L4221-L4230
- PageToken:
https://github.com/apache/iceberg/blob/e7a5a87f26f9de5b200254155aa037368b13a29c/open-api/rest-catalog-open-api.yaml#L2233-L2256

## Are these changes tested?

Yes, includes unit tests for paginated cases.

## Are there any user-facing changes?

- Before: returned incomplete namespace list when server paginated (only
first page)
- After: returns complete namespace list (fetches all pages)

<!-- In the case of user-facing changes, please add the changelog label.
-->

---------

Co-authored-by: Kevin Liu <kevinjqliu@users.noreply.github.com>

Author: ebyhr

pyiceberg/catalog/rest/__init__.py

@@ -350,6 +350,7 @@ def _parse_endpoints(cls, v: list[str] | None) -> set[Endpoint] | None:
 
 class ListNamespaceResponse(IcebergBaseModel):
     namespaces: list[Identifier] = Field()
+    next_page_token: str | None = Field(default=None, alias="next-page-token")
 
 
 class NamespaceResponse(IcebergBaseModel):
@@ -1243,19 +1244,31 @@ def drop_namespace(self, namespace: str | Identifier) -> None:
     def list_namespaces(self, namespace: str | Identifier = ()) -> list[Identifier]:
         self._check_endpoint(Capability.V1_LIST_NAMESPACES)
         namespace_tuple = self.identifier_to_tuple(namespace)
-        response = self._session.get(
-            self.url(
-                f"{Endpoints.list_namespaces}?parent={self._encode_namespace_path(namespace_tuple)}"
-                if namespace_tuple
-                else Endpoints.list_namespaces
-            ),
-        )
-        try:
-            response.raise_for_status()
-        except HTTPError as exc:
-            _handle_non_200_response(exc, {404: NoSuchNamespaceError})
 
-        return ListNamespaceResponse.model_validate_json(response.text).namespaces
+        namespaces: list[Identifier] = []
+        page_token: str | None = None
+
+        while True:
+            params: dict[str, str] = {}
+            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)
+
+            if not parsed.next_page_token:
+                break
+            page_token = parsed.next_page_token
+
+        return namespaces
 
     @retry(**_RETRY_ARGS)
     @override

tests/catalog/test_rest.py

@@ -823,6 +823,105 @@ def test_list_namespace_with_parent_200(rest_mock: Mocker) -> None:
     ]
 
 
+def test_list_namespaces_paginated_200(rest_mock: Mocker) -> None:
+    # First page with next-page-token
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces",
+        json={
+            "namespaces": [["ns1"], ["ns2"]],
+            "next-page-token": "page2token",
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    # Second page with next-page-token
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces?pageToken=page2token",
+        json={
+            "namespaces": [["ns3"], ["ns4"]],
+            "next-page-token": "page3token",
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    # Third page without next-page-token (last page)
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces?pageToken=page3token",
+        json={
+            "namespaces": [["ns5"]],
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    result = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).list_namespaces()
+    assert result == [
+        ("ns1",),
+        ("ns2",),
+        ("ns3",),
+        ("ns4",),
+        ("ns5",),
+    ]
+
+
+def test_list_namespaces_with_parent_paginated_200(rest_mock: Mocker) -> None:
+    # First page
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces?parent=accounting",
+        json={
+            "namespaces": [["accounting", "tax"]],
+            "next-page-token": "page2",
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    # Second page (last)
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces?parent=accounting&pageToken=page2",
+        json={
+            "namespaces": [["accounting", "payroll"]],
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    result = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).list_namespaces(("accounting",))
+    assert result == [
+        ("accounting", "tax"),
+        ("accounting", "payroll"),
+    ]
+
+
+def test_list_namespaces_paginated_200_none_next_page_token(rest_mock: Mocker) -> None:
+    # First page with next-page-token
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces",
+        json={
+            "namespaces": [["ns1"], ["ns2"]],
+            "next-page-token": "page2token",
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    # The last page with None next-page-token
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces?pageToken=page2token",
+        json={
+            "namespaces": [["ns3"]],
+            "next-page-token": None,
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    result = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).list_namespaces()
+    assert result == [
+        ("ns1",),
+        ("ns2",),
+        ("ns3",),
+    ]
+
+
 def test_list_namespace_with_parent_404(rest_mock: Mocker) -> None:
     rest_mock.get(
         f"{TEST_URI}v1/namespaces?parent=some_namespace",