rollback ruff changes

Author: Alexandra Popova

pyiceberg/catalog/rest/__init__.py

@@ -20,6 +20,7 @@
     Any,
     Dict,
     List,
+    Optional,
     Set,
     Tuple,
     Union,
@@ -152,18 +153,18 @@ def _retry_hook(retry_state: RetryCallState) -> None:
 
 
 class TableResponse(IcebergBaseModel):
-    metadata_location: str | None = Field(alias="metadata-location", default=None)
+    metadata_location: Optional[str] = Field(alias="metadata-location", default=None)
     metadata: TableMetadata
     config: Properties = Field(default_factory=dict)
-    storage_credentials: Properties | None = Field(alias="storage-credentials", default=None)
+    storage_credentials: Optional[Properties] = Field(alias="storage-credentials", default=None)
 
 
 class CreateTableRequest(IcebergBaseModel):
     name: str = Field()
-    location: str | None = Field()
+    location: Optional[str] = Field()
     table_schema: Schema = Field(alias="schema")
-    partition_spec: PartitionSpec | None = Field(alias="partition-spec")
-    write_order: SortOrder | None = Field(alias="write-order")
+    partition_spec: Optional[PartitionSpec] = Field(alias="partition-spec")
+    write_order: Optional[SortOrder] = Field(alias="write-order")
     stage_create: bool = Field(alias="stage-create", default=False)
     properties: Dict[str, str] = Field(default_factory=dict)
 
@@ -179,8 +180,8 @@ class RegisterTableRequest(IcebergBaseModel):
 
 
 class ConfigResponse(IcebergBaseModel):
-    defaults: Properties | None = Field(default_factory=dict)
-    overrides: Properties | None = Field(default_factory=dict)
+    defaults: Optional[Properties] = Field(default_factory=dict)
+    overrides: Optional[Properties] = Field(default_factory=dict)
 
 
 class ListNamespaceResponse(IcebergBaseModel):
@@ -294,7 +295,7 @@ def _create_legacy_oauth2_auth_manager(self, session: Session) -> AuthManager:
 
         return AuthManagerFactory.create("legacyoauth2", auth_config)
 
-    def _check_valid_namespace_identifier(self, identifier: str | Identifier) -> Identifier:
+    def _check_valid_namespace_identifier(self, identifier: Union[str, Identifier]) -> Identifier:
         """Check if the identifier has at least one element."""
         identifier_tuple = Catalog.identifier_to_tuple(identifier)
         if len(identifier_tuple) < 1:
@@ -377,22 +378,22 @@ def _fetch_config(self) -> None:
         # Update URI based on overrides
         self.uri = config[URI]
 
-    def _identifier_to_validated_tuple(self, identifier: str | Identifier) -> Identifier:
+    def _identifier_to_validated_tuple(self, identifier: Union[str, Identifier]) -> Identifier:
         identifier_tuple = self.identifier_to_tuple(identifier)
         if len(identifier_tuple) <= 1:
             raise NoSuchIdentifierError(f"Missing namespace or invalid identifier: {'.'.join(identifier_tuple)}")
         return identifier_tuple
 
     def _split_identifier_for_path(
-        self, identifier: str | Identifier | TableIdentifier, kind: IdentifierKind = IdentifierKind.TABLE
+        self, identifier: Union[str, Identifier, TableIdentifier], kind: IdentifierKind = IdentifierKind.TABLE
     ) -> Properties:
         if isinstance(identifier, TableIdentifier):
             return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root), kind.value: identifier.name}
         identifier_tuple = self._identifier_to_validated_tuple(identifier)
 
         return {"namespace": NAMESPACE_SEPARATOR.join(identifier_tuple[:-1]), kind.value: identifier_tuple[-1]}
 
-    def _split_identifier_for_json(self, identifier: str | Identifier) -> Dict[str, Identifier | str]:
+    def _split_identifier_for_json(self, identifier: Union[str, Identifier]) -> Dict[str, Union[Identifier, str]]:
         identifier_tuple = self._identifier_to_validated_tuple(identifier)
         return {"namespace": identifier_tuple[:-1], "name": identifier_tuple[-1]}
 
@@ -496,9 +497,9 @@ def _config_headers(self, session: Session) -> None:
 
     def _create_table(
         self,
-        identifier: str | Identifier,
+        identifier: Union[str, Identifier],
         schema: Union[Schema, "pa.Schema"],
-        location: str | None = None,
+        location: Optional[str] = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
         properties: Properties = EMPTY_DICT,
@@ -538,9 +539,9 @@ def _create_table(
     @retry(**_RETRY_ARGS)
     def create_table(
         self,
-        identifier: str | Identifier,
+        identifier: Union[str, Identifier],
         schema: Union[Schema, "pa.Schema"],
-        location: str | None = None,
+        location: Optional[str] = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
         properties: Properties = EMPTY_DICT,
@@ -559,9 +560,9 @@ def create_table(
     @retry(**_RETRY_ARGS)
     def create_table_transaction(
         self,
-        identifier: str | Identifier,
+        identifier: Union[str, Identifier],
         schema: Union[Schema, "pa.Schema"],
-        location: str | None = None,
+        location: Optional[str] = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
         properties: Properties = EMPTY_DICT,
@@ -579,7 +580,7 @@ def create_table_transaction(
         return CreateTableTransaction(staged_table)
 
     @retry(**_RETRY_ARGS)
-    def register_table(self, identifier: str | Identifier, metadata_location: str) -> Table:
+    def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
         """Register a new table using existing metadata.
 
         Args:
@@ -611,7 +612,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str) -
         return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
 
     @retry(**_RETRY_ARGS)
-    def list_tables(self, namespace: str | Identifier) -> List[Identifier]:
+    def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         namespace_tuple = self._check_valid_namespace_identifier(namespace)
         namespace_concat = NAMESPACE_SEPARATOR.join(namespace_tuple)
         response = self._session.get(self.url(Endpoints.list_tables, namespace=namespace_concat))
@@ -622,7 +623,7 @@ def list_tables(self, namespace: str | Identifier) -> List[Identifier]:
         return [(*table.namespace, table.name) for table in ListTablesResponse.model_validate_json(response.text).identifiers]
 
     @retry(**_RETRY_ARGS)
-    def load_table(self, identifier: str | Identifier) -> Table:
+    def load_table(self, identifier: Union[str, Identifier]) -> Table:
         params = {}
         if mode := self.properties.get(SNAPSHOT_LOADING_MODE):
             if mode in {"all", "refs"}:
@@ -642,7 +643,7 @@ def load_table(self, identifier: str | Identifier) -> Table:
         return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
 
     @retry(**_RETRY_ARGS)
-    def drop_table(self, identifier: str | Identifier, purge_requested: bool = False) -> None:
+    def drop_table(self, identifier: Union[str, Identifier], purge_requested: bool = False) -> None:
         response = self._session.delete(
             self.url(Endpoints.drop_table, prefixed=True, **self._split_identifier_for_path(identifier)),
             params={"purgeRequested": purge_requested},
@@ -653,11 +654,11 @@ def drop_table(self, identifier: str | Identifier, purge_requested: bool = False
             _handle_non_200_response(exc, {404: NoSuchTableError})
 
     @retry(**_RETRY_ARGS)
-    def purge_table(self, identifier: str | Identifier) -> None:
+    def purge_table(self, identifier: Union[str, Identifier]) -> None:
         self.drop_table(identifier=identifier, purge_requested=True)
 
     @retry(**_RETRY_ARGS)
-    def rename_table(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> Table:
+    def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[str, Identifier]) -> Table:
         payload = {
             "source": self._split_identifier_for_json(from_identifier),
             "destination": self._split_identifier_for_json(to_identifier),
@@ -682,7 +683,7 @@ def _remove_catalog_name_from_table_request_identifier(self, table_request: Comm
         return table_request
 
     @retry(**_RETRY_ARGS)
-    def list_views(self, namespace: str | Identifier) -> List[Identifier]:
+    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         namespace_tuple = self._check_valid_namespace_identifier(namespace)
         namespace_concat = NAMESPACE_SEPARATOR.join(namespace_tuple)
         response = self._session.get(self.url(Endpoints.list_views, namespace=namespace_concat))
@@ -739,7 +740,7 @@ def commit_table(
         return CommitTableResponse.model_validate_json(response.text)
 
     @retry(**_RETRY_ARGS)
-    def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
         namespace_tuple = self._check_valid_namespace_identifier(namespace)
         payload = {"namespace": namespace_tuple, "properties": properties}
         response = self._session.post(self.url(Endpoints.create_namespace), json=payload)
@@ -749,7 +750,7 @@ def create_namespace(self, namespace: str | Identifier, properties: Properties =
             _handle_non_200_response(exc, {409: NamespaceAlreadyExistsError})
 
     @retry(**_RETRY_ARGS)
-    def drop_namespace(self, namespace: str | Identifier) -> None:
+    def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         namespace_tuple = self._check_valid_namespace_identifier(namespace)
         namespace = NAMESPACE_SEPARATOR.join(namespace_tuple)
         response = self._session.delete(self.url(Endpoints.drop_namespace, namespace=namespace))
@@ -759,7 +760,7 @@ def drop_namespace(self, namespace: str | Identifier) -> None:
             _handle_non_200_response(exc, {404: NoSuchNamespaceError, 409: NamespaceNotEmptyError})
 
     @retry(**_RETRY_ARGS)
-    def list_namespaces(self, namespace: str | Identifier = ()) -> List[Identifier]:
+    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
         namespace_tuple = self.identifier_to_tuple(namespace)
         response = self._session.get(
             self.url(
@@ -776,7 +777,7 @@ def list_namespaces(self, namespace: str | Identifier = ()) -> List[Identifier]:
         return ListNamespaceResponse.model_validate_json(response.text).namespaces
 
     @retry(**_RETRY_ARGS)
-    def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
+    def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
         namespace_tuple = self._check_valid_namespace_identifier(namespace)
         namespace = NAMESPACE_SEPARATOR.join(namespace_tuple)
         response = self._session.get(self.url(Endpoints.load_namespace_metadata, namespace=namespace))
@@ -789,7 +790,7 @@ def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
 
     @retry(**_RETRY_ARGS)
     def update_namespace_properties(
-        self, namespace: str | Identifier, removals: Set[str] | None = None, updates: Properties = EMPTY_DICT
+        self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = None, updates: Properties = EMPTY_DICT
     ) -> PropertiesUpdateSummary:
         namespace_tuple = self._check_valid_namespace_identifier(namespace)
         namespace = NAMESPACE_SEPARATOR.join(namespace_tuple)
@@ -807,7 +808,7 @@ def update_namespace_properties(
         )
 
     @retry(**_RETRY_ARGS)
-    def namespace_exists(self, namespace: str | Identifier) -> bool:
+    def namespace_exists(self, namespace: Union[str, Identifier]) -> bool:
         namespace_tuple = self._check_valid_namespace_identifier(namespace)
         namespace = NAMESPACE_SEPARATOR.join(namespace_tuple)
         response = self._session.head(self.url(Endpoints.namespace_exists, namespace=namespace))
@@ -825,7 +826,7 @@ def namespace_exists(self, namespace: str | Identifier) -> bool:
         return False
 
     @retry(**_RETRY_ARGS)
-    def table_exists(self, identifier: str | Identifier) -> bool:
+    def table_exists(self, identifier: Union[str, Identifier]) -> bool:
         """Check if a table exists.
 
         Args:
@@ -851,7 +852,7 @@ def table_exists(self, identifier: str | Identifier) -> bool:
         return False
 
     @retry(**_RETRY_ARGS)
-    def view_exists(self, identifier: str | Identifier) -> bool:
+    def view_exists(self, identifier: Union[str, Identifier]) -> bool:
         """Check if a view exists.
 
         Args:
@@ -876,7 +877,7 @@ def view_exists(self, identifier: str | Identifier) -> bool:
         return False
 
     @retry(**_RETRY_ARGS)
-    def drop_view(self, identifier: str) -> None:
+    def drop_view(self, identifier: Union[str]) -> None:
         response = self._session.delete(
             self.url(Endpoints.drop_view, prefixed=True, **self._split_identifier_for_path(identifier, IdentifierKind.VIEW)),
         )
@@ -890,4 +891,4 @@ def close(self) -> None:
 
         This method closes mounted HttpAdapters' pooled connections and any active Proxy pooled connections.
         """
-        self._session.close()
+        self._session.close()