Add list_views to rest catalog (apache#817)

* Add list_views to rest catalog

* Update pyiceberg/catalog/__init__.py

Co-authored-by: Sung Yun <107272191+sungwy@users.noreply.github.com>

---------

Co-authored-by: Sung Yun <107272191+sungwy@users.noreply.github.com>

Author: ndrluis

pyiceberg/catalog/__init__.py

@@ -583,6 +583,22 @@ def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identi
             NoSuchNamespaceError: If a namespace with the given name does not exist.
         """
 
+    @abstractmethod
+    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+        """List views under the given namespace in the catalog.
+
+        If namespace is not provided, lists all views in the catalog.
+
+        Args:
+            namespace (str | Identifier): Namespace identifier to search.
+
+        Returns:
+            List[Identifier]: list of table identifiers.
+
+        Raises:
+            NoSuchNamespaceError: If a namespace with the given name does not exist.
+        """
+
     @abstractmethod
     def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
         """Get properties for a namespace.

pyiceberg/catalog/dynamodb.py

@@ -527,6 +527,9 @@ def update_namespace_properties(
 
         return properties_update_summary
 
+    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+        raise NotImplementedError
+
     def _get_iceberg_table_item(self, database_name: str, table_name: str) -> Dict[str, Any]:
         try:
             return self._get_dynamo_item(identifier=f"{database_name}.{table_name}", namespace=database_name)

pyiceberg/catalog/glue.py

@@ -768,3 +768,6 @@ def update_namespace_properties(
         self.glue.update_database(Name=database_name, DatabaseInput=_construct_database_input(database_name, updated_properties))
 
         return properties_update_summary
+
+    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+        raise NotImplementedError

pyiceberg/catalog/hive.py

@@ -389,6 +389,9 @@ def register_table(self, identifier: Union[str, Identifier], metadata_location:
         """
         raise NotImplementedError
 
+    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+        raise NotImplementedError
+
     def _create_lock_request(self, database_name: str, table_name: str) -> LockRequest:
         lock_component: LockComponent = LockComponent(
             level=LockLevel.TABLE, type=LockType.EXCLUSIVE, dbname=database_name, tablename=table_name, isTransactional=True

pyiceberg/catalog/noop.py

@@ -113,3 +113,6 @@ def update_namespace_properties(
         self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = None, updates: Properties = EMPTY_DICT
     ) -> PropertiesUpdateSummary:
         raise NotImplementedError
+
+    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+        raise NotImplementedError

pyiceberg/catalog/rest.py

@@ -96,6 +96,7 @@ class Endpoints:
     table_exists: str = "namespaces/{namespace}/tables/{table}"
     get_token: str = "oauth/tokens"
     rename_table: str = "tables/rename"
+    list_views: str = "namespaces/{namespace}/views"
 
 
 AUTHORIZATION_HEADER = "Authorization"
@@ -200,10 +201,19 @@ class ListTableResponseEntry(IcebergBaseModel):
     namespace: Identifier = Field()
 
 
+class ListViewResponseEntry(IcebergBaseModel):
+    name: str = Field()
+    namespace: Identifier = Field()
+
+
 class ListTablesResponse(IcebergBaseModel):
     identifiers: List[ListTableResponseEntry] = Field()
 
 
+class ListViewsResponse(IcebergBaseModel):
+    identifiers: List[ListViewResponseEntry] = Field()
+
+
 class ErrorResponseMessage(IcebergBaseModel):
     message: str = Field()
     type: str = Field()
@@ -713,6 +723,17 @@ def _remove_catalog_name_from_table_request_identifier(self, table_request: Comm
             )
         return table_request
 
+    @retry(**_RETRY_ARGS)
+    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+        namespace_tuple = self._check_valid_namespace_identifier(namespace)
+        namespace_concat = NAMESPACE_SEPARATOR.join(namespace_tuple)
+        response = self._session.get(self.url(Endpoints.list_views, namespace=namespace_concat))
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            self._handle_non_200_response(exc, {404: NoSuchNamespaceError})
+        return [(*view.namespace, view.name) for view in ListViewsResponse(**response.json()).identifiers]
+
     @retry(**_RETRY_ARGS)
     def _commit_table(self, table_request: CommitTableRequest) -> CommitTableResponse:
         """Update the table.

pyiceberg/catalog/sql.py

@@ -695,3 +695,6 @@ def update_namespace_properties(
                 session.execute(insert_stmt)
             session.commit()
         return properties_update_summary
+
+    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+        raise NotImplementedError

tests/catalog/test_base.py

@@ -256,6 +256,9 @@ def update_namespace_properties(
             removed=list(removed or []), updated=list(updates.keys() if updates else []), missing=list(expected_to_change)
         )
 
+    def list_views(self, namespace: Optional[Union[str, Identifier]] = None) -> List[Identifier]:
+        raise NotImplementedError
+
 
 @pytest.fixture
 def catalog(tmp_path: PosixPath) -> InMemoryCatalog:

tests/catalog/test_rest.py

@@ -402,6 +402,52 @@ def test_list_tables_404(rest_mock: Mocker) -> None:
     assert "Namespace does not exist" in str(e.value)
 
 
+def test_list_views_200(rest_mock: Mocker) -> None:
+    namespace = "examples"
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/{namespace}/views",
+        json={"identifiers": [{"namespace": ["examples"], "name": "fooshare"}]},
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    assert RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).list_views(namespace) == [("examples", "fooshare")]
+
+
+def test_list_views_200_sigv4(rest_mock: Mocker) -> None:
+    namespace = "examples"
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/{namespace}/views",
+        json={"identifiers": [{"namespace": ["examples"], "name": "fooshare"}]},
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    assert RestCatalog("rest", **{"uri": TEST_URI, "token": TEST_TOKEN, "rest.sigv4-enabled": "true"}).list_views(namespace) == [
+        ("examples", "fooshare")
+    ]
+    assert rest_mock.called
+
+
+def test_list_views_404(rest_mock: Mocker) -> None:
+    namespace = "examples"
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/{namespace}/views",
+        json={
+            "error": {
+                "message": "Namespace does not exist: personal in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+                "type": "NoSuchNamespaceException",
+                "code": 404,
+            }
+        },
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+    with pytest.raises(NoSuchNamespaceError) as e:
+        RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).list_views(namespace)
+    assert "Namespace does not exist" in str(e.value)
+
+
 def test_list_namespaces_200(rest_mock: Mocker) -> None:
     rest_mock.get(
         f"{TEST_URI}v1/namespaces",