pyiceberg/catalog/noop.py

#  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 __future__ import annotations

from typing import (
    TYPE_CHECKING,
)

from typing_extensions import override

from pyiceberg.catalog import Catalog, PropertiesUpdateSummary
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.table import (
    CommitTableResponse,
    CreateTableTransaction,
    Table,
)
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.table.update import (
    TableRequirement,
    TableUpdate,
)
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties
from pyiceberg.view import View
from pyiceberg.view.metadata import ViewVersion

if TYPE_CHECKING:
    import pyarrow as pa


class NoopCatalog(Catalog):
    @override
    def create_table(
        self,
        identifier: str | Identifier,
        schema: Schema | pa.Schema,
        location: str | None = None,
        partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
        sort_order: SortOrder = UNSORTED_SORT_ORDER,
        properties: Properties = EMPTY_DICT,
    ) -> Table:
        raise NotImplementedError

    @override
    def create_table_transaction(
        self,
        identifier: str | Identifier,
        schema: Schema | pa.Schema,
        location: str | None = None,
        partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
        sort_order: SortOrder = UNSORTED_SORT_ORDER,
        properties: Properties = EMPTY_DICT,
    ) -> CreateTableTransaction:
        raise NotImplementedError

    @override
    def load_table(self, identifier: str | Identifier) -> Table:
        raise NotImplementedError

    @override
    def table_exists(self, identifier: str | Identifier) -> bool:
        raise NotImplementedError

    @override
    def register_table(self, identifier: str | Identifier, metadata_location: str, overwrite: bool = False) -> Table:
        """Register a new table using existing metadata.

        Args:
            identifier (Union[str, Identifier]): Table identifier for the table
            metadata_location (str): The location to the metadata
            overwrite (bool): Whether to overwrite the existing table, default False

        Returns:
            Table: The newly registered table

        Raises:
            TableAlreadyExistsError: If the table already exists
        """
        raise NotImplementedError

    @override
    def drop_table(self, identifier: str | Identifier) -> None:
        raise NotImplementedError

    @override
    def supports_server_side_planning(self) -> bool:
        return False

    @override
    def purge_table(self, identifier: str | Identifier) -> None:
        raise NotImplementedError

    @override
    def rename_table(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> Table:
        raise NotImplementedError

    @override
    def commit_table(
        self, table: Table, requirements: tuple[TableRequirement, ...], updates: tuple[TableUpdate, ...]
    ) -> CommitTableResponse:
        raise NotImplementedError

    @override
    def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:
        raise NotImplementedError

    @override
    def drop_namespace(self, namespace: str | Identifier) -> None:
        raise NotImplementedError

    @override
    def list_tables(self, namespace: str | Identifier) -> list[Identifier]:
        raise NotImplementedError

    @override
    def list_namespaces(self, namespace: str | Identifier = ()) -> list[Identifier]:
        raise NotImplementedError

    @override
    def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
        raise NotImplementedError

    @override
    def update_namespace_properties(
        self, namespace: str | Identifier, removals: set[str] | None = None, updates: Properties = EMPTY_DICT
    ) -> PropertiesUpdateSummary:
        raise NotImplementedError

    @override
    def list_views(self, namespace: str | Identifier) -> list[Identifier]:
        raise NotImplementedError

    @override
    def view_exists(self, identifier: str | Identifier) -> bool:
        raise NotImplementedError

    @override
    def namespace_exists(self, namespace: str | Identifier) -> bool:
        raise NotImplementedError

    @override
    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
        raise NotImplementedError

    @override
    def drop_view(self, identifier: str | Identifier) -> None:
        raise NotImplementedError

    @override
    def create_view(
        self,
        identifier: str | Identifier,
        schema: Schema | pa.Schema,
        view_version: ViewVersion,
        location: str | None = None,
        properties: Properties = EMPTY_DICT,
    ) -> View:
        raise NotImplementedError

    @override
    def load_view(self, identifier: str | Identifier) -> View:
        raise NotImplementedError