def rename_table()

in pyiceberg/catalog/dynamodb.py [0:0]
38 lines of code
6 McCabe index (conditional complexity)

    def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[str, Identifier]) -> Table:
        """Rename a fully classified table name.

        This method can only rename Iceberg tables in AWS Glue.

        Args:
            from_identifier: Existing table identifier.
            to_identifier: New table identifier.

        Returns:
            Table: the updated table instance with its metadata.

        Raises:
            ValueError: When from table identifier is invalid.
            NoSuchTableError: When a table with the name does not exist.
            NoSuchIcebergTableError: When from table is not a valid iceberg table.
            NoSuchPropertyException: When from table miss some required properties.
            NoSuchNamespaceError: When the destination namespace doesn't exist.
        """
        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
        to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)

        from_table_item = self._get_iceberg_table_item(database_name=from_database_name, table_name=from_table_name)

        try:
            # Verify that from_identifier is a valid iceberg table
            self._convert_dynamo_table_item_to_iceberg_table(dynamo_table_item=from_table_item)
        except NoSuchPropertyException as e:
            raise NoSuchPropertyException(
                f"Failed to rename table {from_database_name}.{from_table_name} since it is missing required properties"
            ) from e
        except NoSuchIcebergTableError as e:
            raise NoSuchIcebergTableError(
                f"Failed to rename table {from_database_name}.{from_table_name} since it is not a valid iceberg table"
            ) from e

        self._ensure_namespace_exists(database_name=from_database_name)
        self._ensure_namespace_exists(database_name=to_database_name)

        try:
            self._put_dynamo_item(
                item=_get_rename_table_item(
                    from_dynamo_table_item=from_table_item, to_database_name=to_database_name, to_table_name=to_table_name
                ),
                condition_expression=f"attribute_not_exists({DYNAMODB_COL_IDENTIFIER})",
            )
        except ConditionalCheckFailedException as e:
            raise TableAlreadyExistsError(f"Table {to_database_name}.{to_table_name} already exists") from e

        try:
            self.drop_table(from_identifier)
        except (NoSuchTableError, GenericDynamoDbError) as e:
            log_message = f"Failed to drop old table {from_database_name}.{from_table_name}. "

            try:
                self.drop_table(to_identifier)
                log_message += f"Rolled back table creation for {to_database_name}.{to_table_name}."
            except (NoSuchTableError, GenericDynamoDbError):
                log_message += (
                    f"Failed to roll back table creation for {to_database_name}.{to_table_name}. Please clean up manually"
                )

            raise ValueError(log_message) from e

        return self.load_table(to_identifier)