Python HL SDK: use IfNoneMatch to support conditional write (aka exclusive creation)

Closes #7505.

This replaces existing obj.exists() check with conditional writes with 'IfNoneMatch: *'.
I have tried to maintain compat in terms of error raised, by raising `ObjectExistsError` on pre-condition failure.

Author: skshetry

clients/python-wrapper/lakefs/object.py

@@ -6,6 +6,7 @@
 
 import base64
 import binascii
+import http
 import io
 import json
 import os
@@ -399,9 +400,6 @@ def __init__(self,
                  metadata: Optional[dict[str, str]] = None,
                  client: Optional[Client] = None) -> None:
 
-        if 'x' in mode and obj.exists():  # Requires explicit create
-            raise ObjectExistsException
-
         if mode not in get_args(WriteModes):
             raise ValueError(f"invalid write mode: '{mode}'. WriteModes: {WriteModes}")
 
@@ -520,6 +518,8 @@ def _upload_raw(self) -> lakefs_sdk.ObjectStats:
             "Accept": "application/json",
             "Content-Type": self.content_type if self.content_type is not None else "application/octet-stream"
         }
+        if self._mode.startswith("x"):
+            headers["If-None-Match"] = "*"
 
         # Create user metadata headers
         if self.metadata is not None:
@@ -538,6 +538,8 @@ def _upload_raw(self) -> lakefs_sdk.ObjectStats:
                                                                                                headers=headers,
                                                                                                body=self._fd)
 
+        if self._mode.startswith("x") and resp.status == http.HTTPStatus.PRECONDITION_FAILED:
+            raise ObjectExistsException(resp.status, resp.reason, resp.data)
         handle_http_error(resp)
         return lakefs_sdk.ObjectStats(**json.loads(resp.data))
 
@@ -568,10 +570,17 @@ def _upload_presign(self) -> lakefs_sdk.ObjectStats:
                                            checksum=etag,
                                            user_metadata=self.metadata,
                                            content_type=self.content_type)
-        return self._client.sdk_client.staging_api.link_physical_address(self._obj.repo,
-                                                                         self._obj.ref,
-                                                                         self._obj.path,
-                                                                         staging_metadata=staging_metadata)
+        if_none_match = "*" if self._mode.startswith("x") else None
+        try:
+            return self._client.sdk_client.staging_api.link_physical_address(self._obj.repo,
+                                                                             self._obj.ref,
+                                                                             self._obj.path,
+                                                                             staging_metadata=staging_metadata,
+                                                                             if_none_match=if_none_match)
+        except lakefs_sdk.ApiException as e:
+            if self._mode.startswith("x") and e.status == http.HTTPStatus.PRECONDITION_FAILED:
+                raise ObjectExistsException(e.status, e.reason, e.body) from e
+            raise
 
     def readable(self) -> bool:
         """

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

@@ -83,13 +83,14 @@ def test_object_read_seek(setup_repo, pre_sign):
     fd.close()
 
 
-def test_object_upload_exists(setup_repo):
+@pytest.mark.parametrize("pre_sign", (True, False))
+def test_object_upload_exists(setup_repo, pre_sign):
     clt, repo = setup_repo
     data = b"test_data"
     obj = WriteableObject(repository_id=repo.properties.id, reference_id="main", path="test_obj", client=clt).upload(
         data=data)
     with expect_exception_context(ObjectExistsException):
-        obj.upload(data="some_other_data", mode='xb')
+        obj.upload(data="some_other_data", mode='xb', pre_sign=pre_sign)
 
     with obj.reader() as fd:
         assert fd.read() == data