refactor: Take all rpaths as resource names only in transactions

Proof of concept for a single repo-and-branch-scoped transaction.

This ties a transaction to a single repo and branch by taking all file and
directory names by resource only instead of a full URI.

Naturally, this has the subtle side effect that given full URIs are silently
understood as nested paths, and uploaded to the transaction branch without
loud errors or warnings. A section was added to the transaction docs that
details this behavior, but it might be safer to check the input path against
existing repos and branches.

Author: nicholasjng

docs/guides/transactions.md

@@ -20,9 +20,9 @@ from lakefs_spec import LakeFSFileSystem
 fs = LakeFSFileSystem()
 
 with fs.transaction("repo", "main") as tx:
-    fs.put_file("train-data.txt", f"repo/{tx.branch.id}/train-data.txt")
+    fs.put_file("train-data.txt", "train-data.txt")
     tx.commit(message="Add training data")
-    fs.put_file("test-data.txt", f"repo/{tx.branch.id}/test-data.txt")
+    fs.put_file("test-data.txt", "test-data.txt")
     sha = tx.commit(message="Add test data")
     tx.tag(sha, name="My train-test split")
 ```
@@ -35,6 +35,13 @@ The full list of supported lakeFS versioning operations (by default, these opera
 * [`rev_parse`](../reference/lakefs_spec/transaction.md#lakefs_spec.transaction.LakeFSTransaction.rev_parse), for parsing revisions like branch/tag names and SHA fragments into full commit SHAs.
 * [`tag`](../reference/lakefs_spec/transaction.md#lakefs_spec.transaction.LakeFSTransaction.tag), for creating a tag pointing to a commit.
 
+## Limitations of transactions
+
+Transactions are scoped to a single repository and branch only, equal to those given to the `fs.transaction()` context manager.
+When uploading files in a transaction via `fs.put()` or `fs.put_file()`, you **must** give all remote paths as file names.
+If you use a fully qualified URI, leading repository and branch names will be interpreted as subdirectories, which will be created on upload.
+No warnings or errors will be thrown, so be sure to double-check your paths in all transaction scopes.
+
 ## Lifecycle of ephemeral transaction branches
 
 You can control the lifecycle for a transaction branch with the `delete` argument:
@@ -56,7 +63,7 @@ from lakefs_spec import LakeFSFileSystem
 fs = LakeFSFileSystem()
 
 with fs.transaction("repo", "main", delete="onsuccess") as tx:
-    fs.put_file("my-file.txt", f"repo/{tx.branch.id}/my-file.txt")
+    fs.put_file("my-file.txt", "my-file.txt")
     tx.commit(message="Add my-file.txt")
     raise ValueError("oops!")
 ```

src/lakefs_spec/spec.py

@@ -684,6 +684,9 @@ def put_file(
         lpath = stringify_path(lpath)
         rpath = stringify_path(rpath)
 
+        if self._intrans:
+            rpath = self.transaction.make_uri(rpath)
+
         if precheck and Path(lpath).is_file():
             remote_checksum = self.checksum(rpath)
             local_checksum = md5_checksum(lpath, blocksize=self.blocksize)

src/lakefs_spec/transaction.py

@@ -3,6 +3,7 @@
 """
 
 import logging
+import os
 import random
 import string
 import warnings
@@ -145,6 +146,15 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         if self.delete == "always" or (success and self.delete == "onsuccess"):
             self._ephemeral_branch.delete()
 
+    def make_uri(self, path: str | os.PathLike[str]) -> str:
+        spath = str(path)
+        # NB: this fails silently if the input path is already fully qualified.
+        # However, in general, it's impossible to distinguish between a
+        # fully qualified path and a normal nested path, so at most, we
+        # could split off the first segment of the input and check it against existing
+        # repositories.
+        return "/".join([self.repository, self.branch.id, spath])
+
     @property
     def branch(self):
         return self._ephemeral_branch

tests/test_put_file.py

@@ -21,7 +21,7 @@ def test_no_change_postcommit(
     message = f"Add file {random_file.name}"
 
     with fs.transaction(repository, temp_branch) as tx:
-        fs.put(lpath, f"{repository.id}/{tx.branch.id}/{random_file.name}")
+        fs.put(lpath, random_file.name)
         tx.commit(message=message)
 
     commits = list(temp_branch.log(max_amount=2))
@@ -31,7 +31,7 @@ def test_no_change_postcommit(
 
     # put the same file again, this time the diff is empty
     with fs.transaction(repository, temp_branch) as tx:
-        fs.put(lpath, f"{repository.id}/{tx.branch.id}/{random_file.name}", precheck=False)
+        fs.put(lpath, random_file.name, precheck=False)
         tx.commit(message=f"Add file {random_file.name}")
 
     # check that no other commit has happened.

tests/test_transactions.py

@@ -17,12 +17,11 @@ def test_transaction_commit(
     random_file = random_file_factory.make()
 
     lpath = str(random_file)
-    rpath = f"{repository.id}/{temp_branch.id}/{random_file.name}"
 
     message = f"Add file {random_file.name}"
 
     with fs.transaction(repository, temp_branch) as tx:
-        fs.put_file(lpath, f"{repository.id}/{tx.branch.id}/{random_file.name}")
+        fs.put_file(lpath, random_file.name)
         assert len(tx.files) == 1
         # sha is a placeholder for the actual SHA created on transaction completion.
         sha = tx.commit(message=message)
@@ -65,7 +64,7 @@ def test_transaction_merge(
             tbname = tx.branch.id
             lpath = str(random_file)
             # stage a file on the transaction branch...
-            fs.put_file(lpath, f"{repository.id}/{tx.branch.id}/{random_file.name}")
+            fs.put_file(lpath, random_file.name)
             # ... commit it with the above message
             tx.commit(message=message)
             # ... and merge it into temp_branch.
@@ -90,7 +89,7 @@ def test_transaction_revert(
     message = f"Add file {random_file.name}"
 
     with fs.transaction(repository, temp_branch, automerge=True) as tx:
-        fs.put_file(lpath, f"{repository.id}/{tx.branch.id}/{random_file.name}")
+        fs.put_file(lpath, random_file.name)
         tx.commit(message=message)
         revert_commit = tx.revert(temp_branch, temp_branch.head)
 
@@ -113,7 +112,7 @@ def test_transaction_failure(
 
     try:
         with fs.transaction(repository, temp_branch) as tx:
-            fs.put_file(lpath, f"{repository.id}/{tx.branch.id}/{random_file.name}")
+            fs.put_file(lpath, random_file.name)
             tx.commit(message=message)
             raise RuntimeError("something went wrong")
     except RuntimeError:
@@ -159,8 +158,8 @@ def test_warn_uncommitted_changes(
     lpath = str(random_file)
 
     with pytest.warns(match="uncommitted changes.*lost"):
-        with fs.transaction(repository, temp_branch) as tx:
-            fs.put_file(lpath, f"{repository.id}/{tx.branch.id}/{random_file.name}")
+        with fs.transaction(repository, temp_branch):
+            fs.put_file(lpath, random_file.name)
 
 
 def test_warn_uncommitted_changes_on_persisted_branch(
@@ -175,4 +174,4 @@ def test_warn_uncommitted_changes_on_persisted_branch(
 
     with pytest.warns(match="uncommitted changes(?:(?!lost).)*$"):
         with fs.transaction(repository, temp_branch, delete="never") as tx:
-            fs.put_file(lpath, f"{repository.id}/{tx.branch.id}/{random_file.name}")
+            fs.put_file(lpath, random_file.name)