view

View

An Iceberg view.

Source code in pyiceberg/view/__init__.py

class View:
    """An Iceberg view."""

    _identifier: Identifier
    metadata: ViewMetadata

    def __init__(
        self,
        identifier: Identifier,
        metadata: ViewMetadata,
    ) -> None:
        self._identifier = identifier
        self.metadata = metadata

    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)

    @property
    def versions(self) -> list[ViewVersion]:
        """Get the versions of this view."""
        return self.metadata.versions

    def version(self, version_id: int) -> ViewVersion | None:
        """Get the version in this view by ID, or None if the ID cannot be found."""
        return next((version for version in self.metadata.versions if version.version_id == version_id), None)

    def history(self) -> list[ViewHistoryEntry]:
        """Get the version of this history view."""
        return self.metadata.version_log

    @property
    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 | None:
        """Return the view representation for the sql dialect, or None if no representation could be resolved."""
        return next(
            (repr.root for repr in self.current_version().representations if repr.root.dialect.casefold() == dialect.casefold()),
            None,
        )

    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

properties property

Return a map of string properties for this view.

versions property

Get the versions of this view.

__eq__(other)

Return the equality of two instances of the View class.

Source code in pyiceberg/view/__init__.py

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

current_version()

Get the version of this view.

Source code in pyiceberg/view/__init__.py

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)

history()

Get the version of this history view.

Source code in pyiceberg/view/__init__.py

def history(self) -> list[ViewHistoryEntry]:
    """Get the version of this history view."""
    return self.metadata.version_log

location()

Return the view's base location.

Source code in pyiceberg/view/__init__.py

def location(self) -> str:
    """Return the view's base location."""
    return self.metadata.location

name()

Return the identifier of this view.

Source code in pyiceberg/view/__init__.py

def name(self) -> Identifier:
    """Return the identifier of this view."""
    return self._identifier

schema()

Return the schema for this view.

Source code in pyiceberg/view/__init__.py

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)

schemas()

Return the schemas for this view.

Source code in pyiceberg/view/__init__.py

def schemas(self) -> dict[int, Schema]:
    """Return the schemas for this view."""
    return {schema.schema_id: schema for schema in self.metadata.schemas}

sql_for(dialect)

Return the view representation for the sql dialect, or None if no representation could be resolved.

Source code in pyiceberg/view/__init__.py

def sql_for(self, dialect: str) -> SQLViewRepresentation | None:
    """Return the view representation for the sql dialect, or None if no representation could be resolved."""
    return next(
        (repr.root for repr in self.current_version().representations if repr.root.dialect.casefold() == dialect.casefold()),
        None,
    )

uuid()

Return the view's UUID.

Source code in pyiceberg/view/__init__.py

def uuid(self) -> UUID:
    """Return the view's UUID."""
    return UUID(self.metadata.view_uuid)

version(version_id)

Get the version in this view by ID, or None if the ID cannot be found.

Source code in pyiceberg/view/__init__.py

def version(self, version_id: int) -> ViewVersion | None:
    """Get the version in this view by ID, or None if the ID cannot be found."""
    return next((version for version in self.metadata.versions if version.version_id == version_id), None)