Add support for registering views

Author: ebyhr

mkdocs/docs/api.md

@@ -1519,6 +1519,17 @@ catalog = load_catalog("default")
 catalog.view_exists("default.bar")
 ```
 
+## Register a view
+
+To register a view using existing metadata:
+
+```python
+catalog.register_view(
+    identifier="docs_example.bids",
+    metadata_location="s3://warehouse/path/to/metadata.json"
+)
+```
+
 ## Table Statistics Management
 
 Manage table statistics with operations through the `Table` API:

pyiceberg/catalog/__init__.py

@@ -674,6 +674,21 @@ def update_namespace_properties(
             ValueError: If removals and updates have overlapping keys.
         """
 
+    @abstractmethod
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        """Register a new view using existing metadata.
+
+        Args:
+            identifier (Union[str, Identifier]): View identifier for the view
+            metadata_location (str): The location to the metadata
+
+        Returns:
+            View: The newly registered view
+
+        Raises:
+            ViewAlreadyExistsError: If the view already exists
+        """
+
     @abstractmethod
     def drop_view(self, identifier: str | Identifier) -> None:
         """Drop a view.

pyiceberg/catalog/bigquery_metastore.py

@@ -41,6 +41,7 @@
 from pyiceberg.table.update import TableRequirement, TableUpdate
 from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties
 from pyiceberg.utils.config import Config
+from pyiceberg.view import View
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -304,6 +305,9 @@ def register_table(self, identifier: str | Identifier, metadata_location: str) -
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/dynamodb.py

@@ -552,6 +552,9 @@ def create_view(
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/glue.py

@@ -966,6 +966,9 @@ def create_view(
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/hive.py

@@ -847,6 +847,9 @@ def update_namespace_properties(
 
         return PropertiesUpdateSummary(removed=list(removed or []), updated=list(updated or []), missing=list(expected_to_change))
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/noop.py

@@ -131,6 +131,9 @@ def view_exists(self, identifier: str | Identifier) -> bool:
     def namespace_exists(self, namespace: str | Identifier) -> bool:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/rest/__init__.py

@@ -153,6 +153,7 @@ class Endpoints:
     rename_table: str = "tables/rename"
     list_views: str = "namespaces/{namespace}/views"
     create_view: str = "namespaces/{namespace}/views"
+    register_view: str = "namespaces/{namespace}/register-view"
     drop_view: str = "namespaces/{namespace}/views/{view}"
     view_exists: str = "namespaces/{namespace}/views/{view}"
     plan_table_scan: str = "namespaces/{namespace}/tables/{table}/plan"
@@ -181,6 +182,7 @@ class Capability:
 
     V1_LIST_VIEWS = Endpoint(http_method=HttpMethod.GET, path=f"{API_PREFIX}/{Endpoints.list_views}")
     V1_VIEW_EXISTS = Endpoint(http_method=HttpMethod.HEAD, path=f"{API_PREFIX}/{Endpoints.view_exists}")
+    V1_REGISTER_VIEW = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.register_view}")
     V1_DELETE_VIEW = Endpoint(http_method=HttpMethod.DELETE, path=f"{API_PREFIX}/{Endpoints.drop_view}")
     V1_SUBMIT_TABLE_SCAN_PLAN = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.plan_table_scan}")
     V1_TABLE_SCAN_PLAN_TASKS = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.fetch_scan_tasks}")
@@ -318,6 +320,11 @@ class RegisterTableRequest(IcebergBaseModel):
     metadata_location: str = Field(..., alias="metadata-location")
 
 
+class RegisterViewRequest(IcebergBaseModel):
+    name: str
+    metadata_location: str = Field(..., alias="metadata-location")
+
+
 class ConfigResponse(IcebergBaseModel):
     defaults: Properties | None = Field(default_factory=dict)
     overrides: Properties | None = Field(default_factory=dict)
@@ -1312,6 +1319,27 @@ def view_exists(self, identifier: str | Identifier) -> bool:
 
         return False
 
+    @retry(**_RETRY_ARGS)
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        self._check_endpoint(Capability.V1_REGISTER_VIEW)
+        namespace_and_view = self._split_identifier_for_path(identifier)
+        request = RegisterViewRequest(
+            name=self._identifier_to_validated_tuple(identifier)[-1],
+            metadata_location=metadata_location,
+        )
+        serialized_json = request.model_dump_json().encode(UTF8)
+        response = self._session.post(
+            self.url(Endpoints.register_view, namespace=namespace_and_view["namespace"]),
+            data=serialized_json,
+        )
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {409: ViewAlreadyExistsError})
+
+        view_response = ViewResponse.model_validate_json(response.text)
+        return self._response_to_view(self.identifier_to_tuple(identifier), view_response)
+
     @retry(**_RETRY_ARGS)
     def drop_view(self, identifier: str) -> None:
         self._check_endpoint(Capability.V1_DELETE_VIEW)

pyiceberg/catalog/sql.py

@@ -741,6 +741,9 @@ def list_views(self, namespace: str | Identifier) -> list[Identifier]:
     def view_exists(self, identifier: str | Identifier) -> bool:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

tests/catalog/test_rest.py

@@ -104,6 +104,7 @@
     Capability.V1_REGISTER_TABLE,
     Capability.V1_LIST_VIEWS,
     Capability.V1_VIEW_EXISTS,
+    Capability.V1_REGISTER_VIEW,
     Capability.V1_DELETE_VIEW,
     Capability.V1_SUBMIT_TABLE_SCAN_PLAN,
     Capability.V1_TABLE_SCAN_PLAN_TASKS,
@@ -2122,6 +2123,47 @@ def test_table_identifier_in_commit_table_request(
     )
 
 
+def test_register_view_200(rest_mock: Mocker, example_view_metadata_rest_json: dict[str, Any]) -> None:
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/default/register-view",
+        json=example_view_metadata_rest_json,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    actual = catalog.register_view(
+        identifier=("default", "registered_view"), metadata_location="s3://warehouse/database/view/metadata.json"
+    )
+    expected = View(
+        identifier=("default", "registered_view"),
+        metadata=ViewMetadata(**example_view_metadata_rest_json["metadata"]),
+    )
+    assert actual.metadata.model_dump() == expected.metadata.model_dump()
+    assert actual.name() == expected.name()
+
+
+def test_register_view_409(rest_mock: Mocker) -> None:
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/default/register-view",
+        json={
+            "error": {
+                "message": "View already exists: default.view in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+                "type": "AlreadyExistsException",
+                "code": 409,
+            }
+        },
+        status_code=409,
+        request_headers=TEST_HEADERS,
+    )
+
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    with pytest.raises(ViewAlreadyExistsError) as e:
+        catalog.register_view(
+            identifier=("default", "registered_view"), metadata_location="s3://warehouse/database/view/metadata.json"
+        )
+    assert "View already exists" in str(e.value)
+
+
 def test_drop_view_invalid_namespace(rest_mock: Mocker) -> None:
     view = "view"
     with pytest.raises(NoSuchIdentifierError) as e: