ref: factor out s3.object_exists method

CHANGELOG

@@ -1,4 +1,5 @@
 0.5.5
+ - ref: factor out s3.object_exists method
  - ref: factor out s3.compute_checksum method
 0.5.4
  - enh: cache presigned URLs with 10% tolerance

dcor_shared/s3.py

@@ -195,6 +195,19 @@ def make_object_public(bucket_name, object_name, missing_ok=False):
             )
 
 
+def object_exists(bucket_name, object_name):
+    """Check whether an object exists"""
+    s3_client, _, _ = get_s3()
+    try:
+        s3_client.head_object(Bucket=bucket_name, Key=object_name)
+    except (s3_client.exceptions.NoSuchKey,
+            s3_client.exceptions.ClientError):
+        obj_exists = False
+    else:
+        obj_exists = True
+    return obj_exists
+
+
 @functools.lru_cache()
 def require_bucket(bucket_name):
     """Create an S3 bucket if it does not exist yet
@@ -302,14 +315,8 @@ def upload_file(bucket_name, object_name, path, sha256, private=True,
 
     perform_upload = True
     if not override:
-        try:
-            s3_client.head_object(Bucket=bucket_name, Key=object_name)
-        except (s3_client.exceptions.NoSuchKey,
-                s3_client.exceptions.ClientError):
-            object_exists = False
-        else:
-            object_exists = True
-        perform_upload = not object_exists
+        perform_upload = not object_exists(bucket_name=bucket_name,
+                                           object_name=object_name)
 
     if perform_upload:
         s3_bucket.upload_file(Filename=str(path),

tests/test_s3.py

@@ -96,6 +96,27 @@ def test_make_object_public_no_such_key(tmp_path):
                               missing_ok=False)
 
 
+def test_object_exists():
+    path = data_path / "calibration_beads_47.rtdc"
+    bucket_name = f"test-circle-{uuid.uuid4()}"
+    rid = str(uuid.uuid4())
+    object_name = f"resource/{rid[:3]}/{rid[3:6]}/{rid[6:]}"
+    s3.upload_file(
+        bucket_name=bucket_name,
+        object_name=object_name,
+        path=path,
+        sha256=sha256sum(path),
+        private=True)
+
+    assert s3.object_exists(bucket_name=bucket_name,
+                            object_name=object_name)
+    # sanity checks
+    assert not s3.object_exists(bucket_name=bucket_name,
+                                object_name=f"peter/pan-{uuid.uuid4()}")
+    assert not s3.object_exists(bucket_name=f"hansgunter-{uuid.uuid4()}",
+                                object_name=object_name)
+
+
 def test_presigned_url(tmp_path):
     path = data_path / "calibration_beads_47.rtdc"
     bucket_name = f"test-circle-{uuid.uuid4()}"