feat(python): add tag parameter to Branch.transact() (#9334)

Add optional tag parameter to automatically create tags after successful
transaction completion, eliminating manual commit SHA retrieval.

Usage:

```py
  with branch.transact(commit_message='update', tag='v1.0.0') as tx:
      tx.object('file.txt').upload('data')
  # Tag 'v1.0.0' created on merge commit
```

- Tag created only on transaction success (fail-fast validation)
- Empty tag names treated as None
- Tag accessible via tx.tag attribute

Closes: #8200

Author: skshetry

clients/python-wrapper/lakefs/branch.py

@@ -16,11 +16,12 @@
 from lakefs.import_manager import ImportManager
 from lakefs.reference import Reference, ReferenceType, generate_listing
 from lakefs.models import Change, Commit
+from lakefs.tag import Tag
 from lakefs.exceptions import (
     api_exception_handler,
     ConflictException,
     LakeFSException,
-    TransactionException
+    TransactionException,
 )
 
 
@@ -289,7 +290,7 @@ def import_data(self, commit_message: str = "", metadata: Optional[dict] = None)
 
     @contextmanager
     def transact(self, commit_message: str = "", commit_metadata: Optional[Dict] = None,
-                 delete_branch_on_error: bool = True) -> _Transaction:
+                 delete_branch_on_error: bool = True, tag: Optional[str] = None) -> _Transaction:
         """
         Create a transaction for multiple operations.
         Transaction allows for multiple modifications to be performed atomically on a branch,
@@ -323,10 +324,11 @@ def transact(self, commit_message: str = "", commit_metadata: Optional[Dict] = N
         :param commit_message: once the transaction is committed, a commit is created with this message
         :param commit_metadata: user metadata for the transaction commit
         :param delete_branch_on_error: Defaults to True. Ensures ephemeral branch is deleted on error.
+        :param tag: Optional tag name to create after the transaction is successfully completed
         :return: a Transaction object to perform the operations on
         """
         with Transaction(self._repo_id, self._id, commit_message, commit_metadata, delete_branch_on_error,
-                         self._client) as tx:
+                         self._client, tag) as tx:
             yield tx
 
 
@@ -336,10 +338,11 @@ def _get_tx_name() -> str:
         return f"tx-{uuid.uuid4()}"  # Don't rely on source branch name as this might exceed valid branch length
 
     def __init__(self, repository_id: str, branch_id: str, commit_message: str = "",
-                 commit_metadata: Optional[Dict] = None, client: Client = None):
+                 commit_metadata: Optional[Dict] = None, client: Client = None, tag: Optional[str] = None):
         self._commit_message = commit_message
         self._commit_metadata = commit_metadata
         self._source_branch = branch_id
+        self._tag = tag
 
         tx_name = self._get_tx_name()
         self._tx_branch = Branch(repository_id, tx_name, client).create(branch_id, hidden=True)
@@ -382,6 +385,21 @@ def commit_metadata(self, metadata: Optional[Dict]) -> None:
         """
         self._commit_metadata = metadata
 
+    @property
+    def tag(self) -> Optional[str]:
+        """
+        Return the tag name configured for this transaction completion
+        """
+        return self._tag
+
+    @tag.setter
+    def tag(self, tag: str) -> None:
+        """
+        Set the tag name for this transaction completion
+        :param tag: The tag name to create after the transaction is successfully completed
+        """
+        self._tag = tag
+
 
 class Transaction:
     """
@@ -393,7 +411,8 @@ class Transaction:
     """
 
     def __init__(self, repository_id: str, branch_id: str, commit_message: str = "",
-                 commit_metadata: Optional[Dict] = None, delete_branch_on_error: bool = True, client: Client = None):
+                 commit_metadata: Optional[Dict] = None, delete_branch_on_error: bool = True, client: Client = None,
+                 tag: Optional[str] = None):
         self._repo_id = repository_id
         self._commit_message = commit_message
         self._commit_metadata = commit_metadata
@@ -402,10 +421,11 @@ def __init__(self, repository_id: str, branch_id: str, commit_message: str = "",
         self._tx = None
         self._tx_branch = None
         self._cleanup_branch = delete_branch_on_error
+        self._tag = tag
 
     def __enter__(self):
         self._tx = _Transaction(self._repo_id, self._source_branch, self._commit_message, self._commit_metadata,
-                                self._client)
+                                self._client, self._tag)
         self._tx_branch = Branch(self._repo_id, self._tx.id, self._client)
         return self._tx
 
@@ -417,11 +437,15 @@ def __exit__(self, typ, value, traceback) -> bool:
 
         try:
             self._tx_branch.commit(message=self._tx.commit_message, metadata=self._tx.commit_metadata)
-            self._tx_branch.merge_into(self._source_branch, message=f"Merge transaction {self._tx.id} to branch")
+            merge_commit_id = self._tx_branch.merge_into(
+                self._source_branch, message=f"Merge transaction {self._tx.id} to branch")
             self._tx_branch.delete()
-
-            return False
         except LakeFSException as e:
             if self._cleanup_branch:
                 self._tx_branch.delete()
             raise TransactionException(f"Failed committing transaction {self._tx.id}: {e}") from e
+
+        if self._tx.tag:
+            tag = Tag(self._repo_id, self._tx.tag, self._client)
+            tag.create(merge_commit_id)
+        return False

clients/python-wrapper/tests/integration/test_branch.py

@@ -6,7 +6,7 @@
     from pydantic import ValidationError
 
 import lakefs
-from lakefs.exceptions import NotFoundException, TransactionException
+from lakefs.exceptions import NotFoundException, TransactionException, ConflictException
 from tests.utests.common import expect_exception_context
 
 
@@ -129,6 +129,8 @@ def test_transaction(setup_repo):
         tx.reset_changes(path_type="common_prefix", path="foo")
         tx_id = tx.id
 
+    assert not list(repo.tags())
+
     # Verify transaction branch was deleted
     with expect_exception_context(NotFoundException):
         repo.branch(tx.id).get_commit()
@@ -207,3 +209,45 @@ def test_transaction_failure(setup_repo, cleanup_branch):
         assert tx.get_commit()
 
     assert test_branch.get_commit() == new_ref.get_commit()
+
+
+def test_transaction_with_tag(setup_repo):
+    _, repo = setup_repo
+    path_and_data = ["a", "b", "bar/a", "bar/b", "bar/c", "c"]
+    test_branch = repo.branch("main")
+
+    with test_branch.transact(commit_message="my transaction with tag", tag="v1.0.0") as tx:
+        assert tx.tag == "v1.0.0"
+        upload_data(tx, path_and_data)
+    assert repo.tag("v1.0.0").get_commit() == test_branch.get_commit()
+
+
+def test_transaction_with_explicit_none_tag(setup_repo):
+    _, repo = setup_repo
+    path_and_data = ["a", "b", "bar/a", "bar/b", "bar/c", "c"]
+    test_branch = repo.branch("main")
+
+    with test_branch.transact(commit_message="my transaction with tag", tag=None) as tx:
+        assert tx.tag is None
+        upload_data(tx, path_and_data)
+    assert not list(repo.tags())
+
+
+def test_transaction_with_existing_tag(setup_repo):
+    _, repo = setup_repo
+    path_and_data = ["a", "b", "bar/a", "bar/b", "bar/c", "c"]
+    test_branch = repo.branch("main")
+
+    repo: "lakefs.Repository"
+
+    test_branch.object("initial_file").upload("initial content")
+    initial_commit = test_branch.commit("initial commit")
+    repo.tag("v1.0.0").create(initial_commit)
+
+    with expect_exception_context(ConflictException, "tag already exists"):
+        with test_branch.transact(commit_message="my transaction with existing tag", tag="v1.0.0") as tx:
+            upload_data(tx, path_and_data)
+
+    # Verify transaction branch was deleted
+    with expect_exception_context(NotFoundException):
+        repo.branch(tx.id).get_commit()