[python] Improve catalog API type annotations

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

paimon-python/pypaimon/api/rest_api.py

@@ -191,7 +191,7 @@ def list_databases_paged(
         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")
 

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 @@ def list_databases(self) -> List[str]:
         """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 @@ def list_tables(self, database_name: str) -> List[str]:
             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 @@ def alter_database(self, name: str, changes: list):
         )
 
     @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 @@ def drop_table(self, identifier: Union[str, Identifier], ignore_if_not_exists: b
             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 @@ def alter_table(
         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 @@ def supports_version_management(self) -> bool:
         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 @@ def commit_snapshot(
 
         """
 
-    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 @@ def list_partitions_paged(
             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 @@ def get_tag(
             self,
             identifier: Union[str, Identifier],
             tag_name: str,
-    ):
+    ) -> GetTagResponse:
         """Get a tag of a table.
 
         Args:
@@ -396,7 +406,7 @@ def list_tags_paged(
             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 @@ def list_tags_paged(
             "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")

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.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 @@ def __init__(self, catalog_options: Options):
             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 @@ def get_database(self, name: str) -> Database:
         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 @@ def create_database(self, name: str, ignore_if_exists: bool, properties: Optiona
             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 @@ def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade:
 
         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 @@ def _load_system_table(self, identifier: Identifier) -> Table:
             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 @@ def create_table(self, identifier: Union[str, Identifier], schema: 'Schema', ign
             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 @@ def alter_table(
         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 @@ def alter_table(
         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 @@ def rename_table(self, source_identifier: Union[str, Identifier], target_identif
         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 @@ def commit_snapshot(
     ) -> 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 @@ def list_partitions_paged(
             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
 

paimon-python/pypaimon/catalog/jdbc_catalog.py

@@ -31,15 +31,18 @@
     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
 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 @@ def get_database(self, name: str) -> Database:
         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 @@ def create_database(self, name: str, ignore_if_exists: bool, properties: Optiona
         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 @@ def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade:
                 (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 @@ def get_table(self, identifier: Union[str, Identifier]) -> Table:
         )
         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 @@ def create_table(self, identifier: Union[str, Identifier], schema: 'Schema', ign
             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 @@ def drop_table(self, identifier: Union[str, Identifier], ignore_if_not_exists: b
             )
         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 @@ def alter_table(
         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 @@ def alter_table(
                 )
                 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 @@ def get_table_path(self, identifier: Identifier) -> str:
         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(