View object API

pyiceberg/view/__init__.py

@@ -16,12 +16,12 @@
 # under the License.
 from __future__ import annotations
 
-from typing import (
-    Any,
-)
+from typing import Any
+from uuid import UUID
 
+from pyiceberg.schema import Schema
 from pyiceberg.typedef import Identifier
-from pyiceberg.view.metadata import ViewMetadata
+from pyiceberg.view.metadata import SQLViewRepresentation, ViewHistoryEntry, ViewMetadata, ViewVersion
 
 
 class View:
@@ -42,6 +42,46 @@ def name(self) -> Identifier:
         """Return the identifier of this view."""
         return self._identifier
 
+    def schema(self) -> Schema:
+        """Return the schema for this view."""
+        return next(schema for schema in self.metadata.schemas if schema.schema_id == self.current_version().schema_id)
+
+    def schemas(self) -> dict[int, Schema]:
+        """Return the schemas for this view."""
+        return {schema.schema_id: schema for schema in self.metadata.schemas}
+
+    def current_version(self) -> ViewVersion:
+        """Get the version of this view."""
+        return next(version for version in self.metadata.versions if version.version_id == self.metadata.current_version_id)
+
+    def versions(self) -> list[ViewVersion]:
+        """Get the versions of this view."""
+        return self.metadata.versions
+
+    def version(self, version_id: int) -> ViewVersion:
+        """Get the version in this view by ID."""
+        return next(version for version in self.metadata.versions if version.version_id == version_id)
+
+    def history(self) -> list[ViewHistoryEntry]:
+        """Get the version of this history view."""
+        return self.metadata.version_log
+
+    def properties(self) -> dict[str, str]:
+        """Return a map of string properties for this view."""
+        return self.metadata.properties
+
+    def location(self) -> str:
+        """Return the view's base location."""
+        return self.metadata.location
+
+    def uuid(self) -> UUID:
+        """Return the view's UUID."""
+        return UUID(self.metadata.view_uuid)
+
+    def sql_for(self, dialect: str) -> SQLViewRepresentation:
+        """Return the view representation for the sql dialect."""
+        return next(repr.root for repr in self.current_version().representations if repr.root.dialect == dialect)
+
     def __eq__(self, other: Any) -> bool:
         """Return the equality of two instances of the View class."""
         return self.name() == other.name() and self.metadata == other.metadata if isinstance(other, View) else False

tests/conftest.py

@@ -1174,6 +1174,43 @@ def example_view_metadata_v1() -> dict[str, Any]:
     }
 
 
+@pytest.fixture
+def example_view_metadata_v1_multiple_versions() -> dict[str, Any]:
+    return {
+        "view-uuid": "a20125c8-7284-442c-9aea-15fee620737c",
+        "format-version": 1,
+        "location": "s3://bucket/test/location/test_view",
+        "current-version-id": 2,
+        "versions": [
+            {
+                "version-id": 1,
+                "timestamp-ms": 1602638573874,
+                "schema-id": 1,
+                "summary": {},
+                "representations": [{"type": "sql", "sql": "SELECT 1", "dialect": "spark"}],
+                "default-namespace": ["default"],
+            },
+            {
+                "version-id": 2,
+                "timestamp-ms": 1602638573875,
+                "schema-id": 2,
+                "summary": {},
+                "representations": [{"type": "sql", "sql": "SELECT 2", "dialect": "spark"}],
+                "default-namespace": ["default"],
+            },
+        ],
+        "schemas": [
+            {"type": "struct", "schema-id": 1, "fields": [{"id": 1, "name": "a", "required": True, "type": "long"}]},
+            {"type": "struct", "schema-id": 2, "fields": [{"id": 2, "name": "b", "required": True, "type": "string"}]},
+        ],
+        "version-log": [
+            {"timestamp-ms": 1602638573874, "version-id": 1},
+            {"timestamp-ms": 1602638573875, "version-id": 2},
+        ],
+        "properties": {},
+    }
+
+
 @pytest.fixture
 def example_table_metadata_v3() -> dict[str, Any]:
     return EXAMPLE_TABLE_METADATA_V3

tests/test_view.py

@@ -0,0 +1,121 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import Any
+from uuid import UUID
+
+import pytest
+
+from pyiceberg.schema import Schema
+from pyiceberg.view import View
+from pyiceberg.view.metadata import SQLViewRepresentation, ViewHistoryEntry, ViewMetadata, ViewVersion
+
+
+@pytest.fixture
+def view(example_view_metadata_v1: dict[str, Any]) -> View:
+    metadata = ViewMetadata.model_validate(example_view_metadata_v1)
+    return View(("default", "test_view"), metadata)
+
+
+def test_view_schema(view: View) -> None:
+    schema = view.schema()
+    assert isinstance(schema, Schema)
+    assert schema.schema_id == 1
+    assert len(schema.fields) == 3
+    assert schema.find_field("x") is not None
+    assert schema.find_field("y") is not None
+    assert schema.find_field("z") is not None
+
+
+def test_view_schemas(view: View) -> None:
+    schemas = view.schemas()
+    assert isinstance(schemas, dict)
+    assert len(schemas) == 1
+    assert 1 in schemas
+    assert isinstance(schemas[1], Schema)
+
+
+def test_view_current_version(view: View) -> None:
+    version = view.current_version()
+    assert isinstance(version, ViewVersion)
+    assert version.version_id == 1
+    assert version.schema_id == 1
+
+
+def test_view_versions(view: View) -> None:
+    versions = view.versions()
+    assert len(versions) == 1
+    assert isinstance(versions[0], ViewVersion)
+    assert versions[0].version_id == 1
+
+
+def test_view_version_by_id(view: View) -> None:
+    version = view.version(1)
+    assert isinstance(version, ViewVersion)
+    assert version.version_id == 1
+    assert version == view.current_version()
+
+
+def test_view_history(view: View) -> None:
+    history = view.history()
+    assert len(history) == 1
+    assert isinstance(history[0], ViewHistoryEntry)
+    assert history[0].version_id == 1
+    assert history[0].timestamp_ms == 1602638573874
+
+
+def test_view_properties(view: View) -> None:
+    assert view.properties() == {"comment": "this is a test view"}
+
+
+def test_view_location(view: View) -> None:
+    assert view.location() == "s3://bucket/test/location/test_view"
+
+
+def test_view_uuid(view: View) -> None:
+    assert view.uuid() == UUID("a20125c8-7284-442c-9aea-15fee620737c")
+
+
+def test_view_sql_for_dialect(view: View) -> None:
+    repr = view.sql_for("spark")
+    assert isinstance(repr, SQLViewRepresentation)
+    assert repr.dialect == "spark"
+    assert repr.sql == "SELECT * FROM prod.db.table"
+
+
+def test_view_schemas_multiple(example_view_metadata_v1_multiple_versions: dict[str, Any]) -> None:
+    view = View(("default", "test_view"), ViewMetadata.model_validate(example_view_metadata_v1_multiple_versions))
+    schemas = view.schemas()
+    assert len(schemas) == 2
+    assert 1 in schemas
+    assert 2 in schemas
+    assert view.schema().schema_id == 2
+
+
+def test_view_versions_multiple(example_view_metadata_v1_multiple_versions: dict[str, Any]) -> None:
+    view = View(("default", "test_view"), ViewMetadata.model_validate(example_view_metadata_v1_multiple_versions))
+    assert len(view.versions()) == 2
+    assert view.current_version().version_id == 2
+
+
+def test_view_version_unknown_id(view: View) -> None:
+    with pytest.raises(StopIteration):
+        view.version(999)
+
+
+def test_view_sql_for_unknown_dialect(view: View) -> None:
+    with pytest.raises(StopIteration):
+        view.sql_for("trino")