Added: public method signature to the init table file.

Moved: the functions for expiring snapshots to their own class.

Author: ForeverAngry

pyiceberg/table/__init__.py

@@ -119,6 +119,7 @@
     ManageSnapshots,
     UpdateSnapshot,
     _FastAppendFiles,
+    ExpireSnapshots
 )
 from pyiceberg.table.update.spec import UpdateSpec
 from pyiceberg.table.update.statistics import UpdateStatistics
@@ -1078,6 +1079,15 @@ def manage_snapshots(self) -> ManageSnapshots:
            ms.create_tag(snapshot_id1, "Tag_A").create_tag(snapshot_id2, "Tag_B")
         """
         return ManageSnapshots(transaction=Transaction(self, autocommit=True))
+    
+    def expire_snapshots(self) -> ExpireSnapshots:
+        """
+        Shorthand to run expire snapshots by id or by a timestamp.
+
+        Use table.expire_snapshots().<operation>().commit() to run a specific operation.
+        Use table.expire_snapshots().<operation-one>().<operation-two>().commit() to run multiple operations.
+        """
+        return ExpireSnapshots(transaction=Transaction(self, autocommit=True))
 
     def update_statistics(self) -> UpdateStatistics:
         """

pyiceberg/table/update/snapshot.py

@@ -850,6 +850,18 @@ def remove_branch(self, branch_name: str) -> ManageSnapshots:
         """
         return self._remove_ref_snapshot(ref_name=branch_name)
 
+class ExpireSnapshots(UpdateTableMetadata["ExpireSnapshots"]):
+    """
+    Expire snapshots by ID or by timestamp.
+    Use table.expire_snapshots().<operation>().commit() to run a specific operation.
+    Use table.expire_snapshots().<operation-one>().<operation-two>().commit() to run multiple operations.
+    Pending changes are applied on commit.
+    """
+
+    _snapshot_ids_to_expire = set()
+    _updates: Tuple[TableUpdate, ...] = ()
+    _requirements: Tuple[TableRequirement, ...] = ()
+
     def _commit(self) -> UpdatesAndRequirements:
         """
         Commit the staged updates and requirements.
@@ -863,7 +875,7 @@ def _commit(self) -> UpdatesAndRequirements:
         self._updates += (update,)
         return self._updates, self._requirements
 
-    def expire_snapshot_by_id(self, snapshot_id: int) -> ManageSnapshots:
+    def expire_snapshot_by_id(self, snapshot_id: int) -> ExpireSnapshots:
         """
         Expire a snapshot by its ID.
 
@@ -878,7 +890,7 @@ def expire_snapshot_by_id(self, snapshot_id: int) -> ManageSnapshots:
         self._snapshot_ids_to_expire.add(snapshot_id)
         return self
 
-    def expire_snapshots_older_than(self, timestamp_ms: int) -> ManageSnapshots:
+    def expire_snapshots_older_than(self, timestamp_ms: int) -> ExpireSnapshots:
         """
         Expire snapshots older than the given timestamp.
 

tests/table/test_expire_snapshots.py

@@ -28,7 +28,7 @@ def test_expire_snapshot(table_v2: Table) -> None:
 
     # Expire the snapshot directly without using a transaction
     try:
-        table_v2.manage_snapshots().expire_snapshot_by_id(EXPIRE_SNAPSHOT).commit()
+        table_v2.expire_snapshots().expire_snapshot_by_id(EXPIRE_SNAPSHOT).commit()
     except Exception as e:
         assert False, f"Commit failed with error: {e}"