This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8c607a584d [python] Improve catalog API type annotations (#10025)
8c607a584d is described below
commit 8c607a584df6b62c4ca2ebd8ccdccd3958ef32ac
Author: zhigang <[email protected]>
AuthorDate: Mon Sep 21 10:39:39 2026 +0800
[python] Improve catalog API type annotations (#10025)
---
paimon-python/pypaimon/api/api_request.py | 2 +-
paimon-python/pypaimon/api/rest_api.py | 2 +-
paimon-python/pypaimon/catalog/catalog.py | 44 +++++++++++++---------
.../pypaimon/catalog/filesystem_catalog.py | 30 ++++++++-------
paimon-python/pypaimon/catalog/jdbc_catalog.py | 24 +++++++-----
.../pypaimon/catalog/rest/rest_catalog.py | 44 +++++++++++++---------
6 files changed, 86 insertions(+), 60 deletions(-)
diff --git a/paimon-python/pypaimon/api/api_request.py
b/paimon-python/pypaimon/api/api_request.py
index 15c9372ac0..8d17642f7a 100644
--- a/paimon-python/pypaimon/api/api_request.py
+++ b/paimon-python/pypaimon/api/api_request.py
@@ -41,7 +41,7 @@ class CreateDatabaseRequest(RESTRequest):
FIELD_OPTIONS = "options"
name: str = json_field(FIELD_NAME)
- options: Dict[str, str] = json_field(FIELD_OPTIONS)
+ options: Optional[Dict[str, str]] = json_field(FIELD_OPTIONS)
@dataclass
diff --git a/paimon-python/pypaimon/api/rest_api.py
b/paimon-python/pypaimon/api/rest_api.py
index 990536cb05..aff1e8793b 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -191,7 +191,7 @@ class RESTApi:
databases = response.data() or []
return PagedList(databases, response.get_next_page_token())
- def create_database(self, name: str, properties: Dict[str, str]) -> None:
+ def create_database(self, name: str, properties: Optional[Dict[str, str]])
-> None:
if not name or not name.strip():
raise ValueError("Database name cannot be empty")
diff --git a/paimon-python/pypaimon/catalog/catalog.py
b/paimon-python/pypaimon/catalog/catalog.py
index ea00ad6dd6..958e84b054 100644
--- a/paimon-python/pypaimon/catalog/catalog.py
+++ b/paimon-python/pypaimon/catalog/catalog.py
@@ -18,11 +18,18 @@
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Union
+from pypaimon.api.api_response import GetTagResponse, PagedList, Partition
+from pypaimon.catalog.database import Database
+from pypaimon.catalog.rest.property_change import PropertyChange
+from pypaimon.catalog.table_query_auth import TableQueryAuthResult
from pypaimon.common.identifier import Identifier
from pypaimon.schema.schema import Schema
from pypaimon.schema.schema_change import SchemaChange
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
+from pypaimon.snapshot.table_snapshot import TableSnapshot
+from pypaimon.table.instant import Instant
+from pypaimon.table.table import Table
class Catalog(ABC):
@@ -43,15 +50,16 @@ class Catalog(ABC):
"""List all database names in the catalog."""
@abstractmethod
- def get_database(self, name: str) -> 'Database':
+ def get_database(self, name: str) -> Database:
"""Get paimon database identified by the given name."""
@abstractmethod
- def create_database(self, name: str, ignore_if_exists: bool, properties:
Optional[dict] = None):
+ def create_database(self, name: str, ignore_if_exists: bool,
+ properties: Optional[Dict[str, str]] = None) -> None:
"""Create a database with properties."""
@abstractmethod
- def drop_database(self, name: str, ignore_if_not_exists: bool = False,
cascade: bool = False):
+ def drop_database(self, name: str, ignore_if_not_exists: bool = False,
cascade: bool = False) -> None:
"""Drop a database.
Args:
@@ -71,7 +79,7 @@ class Catalog(ABC):
List of table names.
"""
- def alter_database(self, name: str, changes: list):
+ def alter_database(self, name: str, changes: List[PropertyChange]) -> None:
"""Alter database properties.
Args:
@@ -86,15 +94,15 @@ class Catalog(ABC):
)
@abstractmethod
- def get_table(self, identifier: Union[str, Identifier]) -> 'Table':
+ def get_table(self, identifier: Union[str, Identifier]) -> Table:
"""Get paimon table identified by the given Identifier."""
@abstractmethod
- def create_table(self, identifier: Union[str, Identifier], schema: Schema,
ignore_if_exists: bool):
+ def create_table(self, identifier: Union[str, Identifier], schema: Schema,
ignore_if_exists: bool) -> None:
"""Create table with schema."""
@abstractmethod
- def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False):
+ def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False) -> None:
"""Drop a table from the catalog.
Args:
@@ -105,7 +113,8 @@ class Catalog(ABC):
TableNotExistException: If table does not exist and
ignore_if_not_exists is False
"""
- def rename_table(self, source_identifier: Union[str, Identifier],
target_identifier: Union[str, Identifier]):
+ def rename_table(self, source_identifier: Union[str, Identifier],
+ target_identifier: Union[str, Identifier]) -> None:
"""Rename a table.
Args:
@@ -125,7 +134,7 @@ class Catalog(ABC):
identifier: Union[str, Identifier],
changes: List[SchemaChange],
ignore_if_not_exists: bool = False
- ):
+ ) -> None:
"""Alter table with schema changes."""
def supports_version_management(self) -> bool:
@@ -138,14 +147,14 @@ class Catalog(ABC):
return False
@abstractmethod
- def load_snapshot(self, identifier: Identifier):
+ def load_snapshot(self, identifier: Identifier) -> Optional[TableSnapshot]:
"""Load the snapshot of table identified by the given Identifier.
Args:
identifier: Path of the table
Returns:
- TableSnapshot instance
+ TableSnapshot instance, or None if no snapshot exists.
Raises:
NotImplementedError: If the catalog does not support version
management
@@ -176,11 +185,12 @@ class Catalog(ABC):
"""
- def rollback_to(self, identifier, instant, from_snapshot=None):
+ def rollback_to(self, identifier: Union[str, Identifier], instant: Instant,
+ from_snapshot: Optional[int] = None) -> None:
"""Rollback table by the given identifier and instant.
Args:
- identifier: Path of the table (Identifier instance).
+ identifier: Path of the table (Identifier or string).
instant: The Instant (SnapshotInstant or TagInstant) to rollback
to.
from_snapshot: Optional snapshot ID. Success only occurs when the
latest snapshot is this snapshot.
@@ -219,7 +229,7 @@ class Catalog(ABC):
max_results: Optional[int] = None,
page_token: Optional[str] = None,
partition_name_pattern: Optional[str] = None,
- ):
+ ) -> PagedList[Partition]:
"""List partitions of a table with pagination.
Args:
@@ -373,7 +383,7 @@ class Catalog(ABC):
self,
identifier: Union[str, Identifier],
tag_name: str,
- ):
+ ) -> GetTagResponse:
"""Get a tag of a table.
Args:
@@ -396,7 +406,7 @@ class Catalog(ABC):
max_results: Optional[int] = None,
page_token: Optional[str] = None,
tag_name_prefix: Optional[str] = None,
- ):
+ ) -> PagedList[str]:
"""List tags of a table with pagination.
Args:
@@ -415,5 +425,5 @@ class Catalog(ABC):
"list_tags_paged is not supported by this catalog."
)
- def auth_table_query(self, identifier: Identifier, select:
Optional[List[str]]) -> 'TableQueryAuthResult':
+ def auth_table_query(self, identifier: Identifier, select:
Optional[List[str]]) -> TableQueryAuthResult:
raise NotImplementedError("auth_table_query not supported by this
catalog")
diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py
b/paimon-python/pypaimon/catalog/filesystem_catalog.py
index be5f7e972f..ccd5b83296 100644
--- a/paimon-python/pypaimon/catalog/filesystem_catalog.py
+++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py
@@ -17,7 +17,7 @@
from typing import Dict, List, Optional, Union
-from pypaimon.api.api_response import GetTagResponse, PagedList
+from pypaimon.api.api_response import GetTagResponse, PagedList, Partition
from pypaimon.catalog.catalog import Catalog
from pypaimon.catalog.catalog_context import CatalogContext
from pypaimon.catalog.catalog_environment import CatalogEnvironment
@@ -38,10 +38,13 @@ from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.time_utils import (duration_to_iso8601,
local_datetime_to_system_zone_millis)
from pypaimon.filesystem.caching_file_io import CachingFileIO
+from pypaimon.schema.schema import Schema
from pypaimon.schema.schema_change import SchemaChange
from pypaimon.schema.schema_manager import SchemaManager
+from pypaimon.schema.table_schema import TableSchema
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
+from pypaimon.snapshot.table_snapshot import TableSnapshot
from pypaimon.table.file_store_table import FileStoreTable
from pypaimon.table.table import Table
@@ -58,7 +61,7 @@ class FileSystemCatalog(Catalog):
FileIO.get(self.warehouse, self.catalog_options),
self.catalog_options,
self._cache_manager)
- def list_databases(self) -> list:
+ def list_databases(self) -> List[str]:
statuses = self.file_io.list_status(self.warehouse)
database_names = []
for status in statuses:
@@ -75,7 +78,8 @@ class FileSystemCatalog(Catalog):
else:
raise DatabaseNotExistException(name)
- def create_database(self, name: str, ignore_if_exists: bool, properties:
Optional[dict] = None):
+ def create_database(self, name: str, ignore_if_exists: bool,
+ properties: Optional[Dict[str, str]] = None) -> None:
try:
self.get_database(name)
if not ignore_if_exists:
@@ -86,7 +90,7 @@ class FileSystemCatalog(Catalog):
path = self.get_database_path(name)
self.file_io.mkdirs(path)
- def drop_database(self, name: str, ignore_if_not_exists: bool = False,
cascade: bool = False):
+ def drop_database(self, name: str, ignore_if_not_exists: bool = False,
cascade: bool = False) -> None:
try:
self.get_database(name)
except DatabaseNotExistException:
@@ -111,7 +115,7 @@ class FileSystemCatalog(Catalog):
self.file_io.delete(db_path, True)
- def list_tables(self, database_name: str) -> list:
+ def list_tables(self, database_name: str) -> List[str]:
try:
self.get_database(database_name)
except DatabaseNotExistException:
@@ -169,7 +173,7 @@ class FileSystemCatalog(Catalog):
raise TableNotExistException(identifier)
return sys_table
- def create_table(self, identifier: Union[str, Identifier], schema:
'Schema', ignore_if_exists: bool):
+ def create_table(self, identifier: Union[str, Identifier], schema: Schema,
ignore_if_exists: bool) -> None:
if schema.options and
schema.options.get(CoreOptions.AUTO_CREATE.key()):
raise ValueError(f"The value of {CoreOptions.AUTO_CREATE.key()}
property should be False.")
@@ -188,7 +192,7 @@ class FileSystemCatalog(Catalog):
schema_manager = SchemaManager(self.file_io, table_path)
schema_manager.create_table(schema)
- def get_table_schema(self, identifier: Identifier):
+ def get_table_schema(self, identifier: Identifier) -> TableSchema:
table_path = self.get_table_path(identifier)
table_schema = SchemaManager(
self.file_io,
@@ -212,7 +216,7 @@ class FileSystemCatalog(Catalog):
identifier: Union[str, Identifier],
changes: List[SchemaChange],
ignore_if_not_exists: bool = False
- ):
+ ) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
@@ -233,7 +237,8 @@ class FileSystemCatalog(Catalog):
except Exception as e:
raise RuntimeError(f"Failed to alter table
{identifier.get_full_name()}: {e}") from e
- def rename_table(self, source_identifier: Union[str, Identifier],
target_identifier: Union[str, Identifier]):
+ def rename_table(self, source_identifier: Union[str, Identifier],
+ target_identifier: Union[str, Identifier]) -> None:
if not isinstance(source_identifier, Identifier):
source_identifier = Identifier.from_string(source_identifier)
if not isinstance(target_identifier, Identifier):
@@ -259,7 +264,7 @@ class FileSystemCatalog(Catalog):
target_path = self.get_table_path(target_identifier)
self.file_io.rename(source_path, target_path)
- def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False):
+ def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
@@ -285,7 +290,7 @@ class FileSystemCatalog(Catalog):
) -> bool:
raise NotImplementedError("This catalog does not support commit
catalog")
- def load_snapshot(self, identifier: Identifier):
+ def load_snapshot(self, identifier: Identifier) -> Optional[TableSnapshot]:
raise NotImplementedError("Filesystem catalog does not support
load_snapshot")
def list_partitions_paged(
@@ -294,8 +299,7 @@ class FileSystemCatalog(Catalog):
max_results: Optional[int] = None,
page_token: Optional[str] = None,
partition_name_pattern: Optional[str] = None,
- ):
- from pypaimon.api.api_response import PagedList, Partition
+ ) -> PagedList[Partition]:
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
from pypaimon.manifest.manifest_list_manager import ManifestListManager
diff --git a/paimon-python/pypaimon/catalog/jdbc_catalog.py
b/paimon-python/pypaimon/catalog/jdbc_catalog.py
index 5ab4a41702..6133b64751 100644
--- a/paimon-python/pypaimon/catalog/jdbc_catalog.py
+++ b/paimon-python/pypaimon/catalog/jdbc_catalog.py
@@ -31,6 +31,7 @@ from pypaimon.catalog.catalog_exception import (
TableNotExistException
)
from pypaimon.catalog.database import Database
+from pypaimon.catalog.rest.property_change import PropertyChange
from pypaimon.common.file_io import FileIO
from pypaimon.common.identifier import Identifier
from pypaimon.common.options.config import CatalogOptions, JdbcCatalogOptions
@@ -38,8 +39,10 @@ from pypaimon.common.options.core_options import CoreOptions
from pypaimon.schema.schema import Schema
from pypaimon.schema.schema_change import SchemaChange
from pypaimon.schema.schema_manager import SchemaManager
+from pypaimon.schema.table_schema import TableSchema
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
+from pypaimon.snapshot.table_snapshot import TableSnapshot
from pypaimon.table.file_store_table import FileStoreTable
from pypaimon.table.table import Table
@@ -314,7 +317,8 @@ class JdbcCatalog(Catalog):
properties.pop(self.DATABASE_EXISTS_PROPERTY, None)
return Database(name, properties)
- def create_database(self, name: str, ignore_if_exists: bool, properties:
Optional[dict] = None):
+ def create_database(self, name: str, ignore_if_exists: bool,
+ properties: Optional[Dict[str, str]] = None) -> None:
if self._database_exists(name):
if not ignore_if_exists:
raise DatabaseAlreadyExistException(name)
@@ -327,7 +331,7 @@ class JdbcCatalog(Catalog):
with self.connection.transaction():
self._insert_database_properties(name, create_props)
- def drop_database(self, name: str, ignore_if_not_exists: bool = False,
cascade: bool = False):
+ def drop_database(self, name: str, ignore_if_not_exists: bool = False,
cascade: bool = False) -> None:
if not self._database_exists(name):
if not ignore_if_not_exists:
raise DatabaseNotExistException(name)
@@ -352,9 +356,8 @@ class JdbcCatalog(Catalog):
(self.catalog_key, name)
)
- def alter_database(self, name: str, changes: list):
+ def alter_database(self, name: str, changes: List[PropertyChange]) -> None:
self.get_database(name)
- from pypaimon.catalog.rest.property_change import PropertyChange
set_properties, remove_keys =
PropertyChange.get_set_properties_to_remove_keys(changes)
current = self._fetch_database_properties(name)
with self.connection.transaction():
@@ -407,7 +410,7 @@ class JdbcCatalog(Catalog):
)
return FileStoreTable(self.file_io, identifier, table_path,
table_schema, catalog_environment)
- def create_table(self, identifier: Union[str, Identifier], schema:
'Schema', ignore_if_exists: bool):
+ def create_table(self, identifier: Union[str, Identifier], schema: Schema,
ignore_if_exists: bool) -> None:
if schema.options and
schema.options.get(CoreOptions.AUTO_CREATE.key()):
raise ValueError(f"The value of {CoreOptions.AUTO_CREATE.key()}
property should be False.")
if not isinstance(identifier, Identifier):
@@ -436,7 +439,7 @@ class JdbcCatalog(Catalog):
self.file_io.delete_directory_quietly(table_path)
raise
- def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False):
+ def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
if not self._table_exists(identifier):
@@ -455,7 +458,8 @@ class JdbcCatalog(Catalog):
)
self.file_io.delete_directory_quietly(table_path)
- def rename_table(self, source_identifier: Union[str, Identifier],
target_identifier: Union[str, Identifier]):
+ def rename_table(self, source_identifier: Union[str, Identifier],
+ target_identifier: Union[str, Identifier]) -> None:
if not isinstance(source_identifier, Identifier):
source_identifier = Identifier.from_string(source_identifier)
if not isinstance(target_identifier, Identifier):
@@ -506,7 +510,7 @@ class JdbcCatalog(Catalog):
identifier: Union[str, Identifier],
changes: List[SchemaChange],
ignore_if_not_exists: bool = False
- ):
+ ) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
if not self._table_exists(identifier):
@@ -524,7 +528,7 @@ class JdbcCatalog(Catalog):
)
self._insert_table_properties(identifier,
self._collect_table_properties(table_schema))
- def get_table_schema(self, identifier: Identifier):
+ def get_table_schema(self, identifier: Identifier) -> TableSchema:
table_schema = SchemaManager(self.file_io,
self.get_table_path(identifier)).latest()
if table_schema is None:
raise TableNotExistException(identifier)
@@ -538,7 +542,7 @@ class JdbcCatalog(Catalog):
db_path = self.get_database_path(identifier.get_database_name())
return f"{db_path}/{identifier.get_table_name()}"
- def load_snapshot(self, identifier: Identifier):
+ def load_snapshot(self, identifier: Identifier) -> Optional[TableSnapshot]:
raise NotImplementedError("JDBC catalog does not support
load_snapshot")
def commit_snapshot(
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index fd325711de..a572c947f9 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -17,7 +17,7 @@
import logging
from typing import Any, Callable, Dict, List, Optional, Union
-from pypaimon.api.api_response import GetTableResponse, PagedList,
ErrorResponse
+from pypaimon.api.api_response import ErrorResponse, GetTableResponse,
GetTagResponse, PagedList, Partition
from pypaimon.api.rest_api import RESTApi
from pypaimon.catalog.catalog_exception import IllegalArgumentError,
IllegalStateError
from pypaimon.api.rest_exception import (NoSuchResourceException,
AlreadyExistsException,
@@ -39,6 +39,7 @@ from pypaimon.catalog.database import Database
from pypaimon.catalog.rest.property_change import PropertyChange
from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO
from pypaimon.catalog.rest.table_metadata import TableMetadata
+from pypaimon.catalog.table_query_auth import TableQueryAuthResult
from pypaimon.common.options.config import CatalogOptions, FuseOptions
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.file_io import FileIO
@@ -49,10 +50,14 @@ from pypaimon.schema.schema_change import SchemaChange
from pypaimon.schema.table_schema import TableSchema
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
+from pypaimon.snapshot.table_snapshot import TableSnapshot
from pypaimon.table.file_store_table import FileStoreTable
from pypaimon.table.format.format_table import FormatTable, Format
from pypaimon.table.iceberg.iceberg_table import IcebergTable
+from pypaimon.table.instant import Instant
from pypaimon.table.object.object_table import ObjectTable
+from pypaimon.table.table import Table
+
logger = logging.getLogger(__name__)
@@ -144,7 +149,8 @@ class RESTCatalog(Catalog):
database_name_pattern: Optional[str] = None) ->
PagedList[str]:
return self.rest_api.list_databases_paged(max_results, page_token,
database_name_pattern)
- def create_database(self, name: str, ignore_if_exists: bool, properties:
Dict[str, str] = None):
+ def create_database(self, name: str, ignore_if_exists: bool,
+ properties: Optional[Dict[str, str]] = None) -> None:
try:
self.rest_api.create_database(name, properties)
except AlreadyExistsException as e:
@@ -167,7 +173,7 @@ class RESTCatalog(Catalog):
except ForbiddenException as e:
raise DatabaseNoPermissionException(name) from e
- def drop_database(self, name: str, ignore_if_not_exists: bool = False,
cascade: bool = False):
+ def drop_database(self, name: str, ignore_if_not_exists: bool = False,
cascade: bool = False) -> None:
if not cascade:
try:
tables = self.list_tables(name)
@@ -189,7 +195,7 @@ class RESTCatalog(Catalog):
except ForbiddenException as e:
raise DatabaseNoPermissionException(name) from e
- def alter_database(self, name: str, changes: List[PropertyChange]):
+ def alter_database(self, name: str, changes: List[PropertyChange]) -> None:
try:
set_properties, remove_keys =
PropertyChange.get_set_properties_to_remove_keys(changes)
self.rest_api.alter_database(name, list(remove_keys),
set_properties)
@@ -227,14 +233,14 @@ class RESTCatalog(Catalog):
except ForbiddenException as e:
raise DatabaseNoPermissionException(database_name) from e
- def get_table(self, identifier: Union[str, Identifier]):
+ def get_table(self, identifier: Union[str, Identifier]) -> Table:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
if identifier.is_system_table():
return self._load_system_table(identifier)
return self._load_data_table(identifier)
- def _load_data_table(self, identifier: Identifier):
+ def _load_data_table(self, identifier: Identifier) -> Union[FormatTable,
IcebergTable, ObjectTable, FileStoreTable]:
return self.load_table(
identifier,
lambda path: self.file_io_for_data(path, identifier),
@@ -242,7 +248,7 @@ class RESTCatalog(Catalog):
self.load_table_metadata,
)
- def _load_system_table(self, identifier: Identifier):
+ def _load_system_table(self, identifier: Identifier) -> Table:
from pypaimon.table.system import system_table_loader
base_identifier = Identifier.create(
@@ -258,7 +264,7 @@ class RESTCatalog(Catalog):
raise TableNotExistException(identifier)
return sys_table
- def create_table(self, identifier: Union[str, Identifier], schema: Schema,
ignore_if_exists: bool):
+ def create_table(self, identifier: Union[str, Identifier], schema: Schema,
ignore_if_exists: bool) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
@@ -267,7 +273,8 @@ class RESTCatalog(Catalog):
if not ignore_if_exists:
raise TableAlreadyExistException(identifier) from e
- def rename_table(self, source_identifier: Union[str, Identifier],
target_identifier: Union[str, Identifier]):
+ def rename_table(self, source_identifier: Union[str, Identifier],
+ target_identifier: Union[str, Identifier]) -> None:
if not isinstance(source_identifier, Identifier):
source_identifier = Identifier.from_string(source_identifier)
if not isinstance(target_identifier, Identifier):
@@ -281,7 +288,7 @@ class RESTCatalog(Catalog):
except ForbiddenException as e:
raise TableNoPermissionException(source_identifier) from e
- def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False):
+ def drop_table(self, identifier: Union[str, Identifier],
ignore_if_not_exists: bool = False) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
@@ -342,7 +349,7 @@ class RESTCatalog(Catalog):
max_results: Optional[int] = None,
page_token: Optional[str] = None,
partition_name_pattern: Optional[str] = None,
- ):
+ ) -> PagedList[Partition]:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
@@ -362,7 +369,7 @@ class RESTCatalog(Catalog):
identifier: Union[str, Identifier],
changes: List[SchemaChange],
ignore_if_not_exists: bool = False
- ):
+ ) -> None:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
@@ -373,7 +380,8 @@ class RESTCatalog(Catalog):
except ForbiddenException as e:
raise TableNoPermissionException(identifier) from e
- def rollback_to(self, identifier, instant, from_snapshot=None):
+ def rollback_to(self, identifier: Union[str, Identifier], instant: Instant,
+ from_snapshot: Optional[int] = None) -> None:
"""Rollback table by the given identifier and instant.
Args:
@@ -401,7 +409,7 @@ class RESTCatalog(Catalog):
except ForbiddenException as e:
raise TableNoPermissionException(identifier) from e
- def load_snapshot(self, identifier: Union[str, Identifier]) ->
Optional['TableSnapshot']:
+ def load_snapshot(self, identifier: Union[str, Identifier]) ->
Optional[TableSnapshot]:
"""Load the latest snapshot for table.
Args:
@@ -547,7 +555,7 @@ class RESTCatalog(Catalog):
except BadRequestException as e:
raise IllegalArgumentError(str(e)) from e
- def get_tag(self, identifier: Union[str, Identifier], tag_name: str):
+ def get_tag(self, identifier: Union[str, Identifier], tag_name: str) ->
GetTagResponse:
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
@@ -695,7 +703,7 @@ class RESTCatalog(Catalog):
internal_file_io: Callable[[str], Any],
external_file_io: Callable[[str], Any],
metadata_loader: Callable[[Identifier], TableMetadata],
- ):
+ ) -> Union[FormatTable, IcebergTable, ObjectTable,
FileStoreTable]:
metadata = metadata_loader(identifier)
schema = metadata.schema
table_type = schema.options.get(CoreOptions.TYPE.key(),
"").strip().lower()
@@ -788,8 +796,8 @@ class RESTCatalog(Catalog):
"""Create FileStoreTable with dynamic options and catalog
environment"""
return FileStoreTable(file_io, catalog_environment.identifier,
table_path, table_schema, catalog_environment)
- def auth_table_query(self, identifier, select=None):
- from pypaimon.catalog.table_query_auth import TableQueryAuthResult
+ def auth_table_query(self, identifier: Identifier,
+ select: Optional[List[str]] = None) ->
TableQueryAuthResult:
try:
response = self.rest_api.auth_table_query(identifier, select)
return TableQueryAuthResult(response.filter,
response.column_masking)
