check for blob existence in registry before uploading

Signed-off-by: Michael Kopf <[email protected]>

CHANGELOG.md

@@ -14,6 +14,7 @@ and **Merged pull requests**. Critical items to know are:
 The versions coincide with releases on pip. Only major versions will be released as tags on Github.
 
 ## [0.0.x](https://github.com/oras-project/oras-py/tree/main) (0.0.x)
+ - check for blob existence before uploading (0.2.26)
  - retry on 500 (0.2.25)
  - align provider config_path type annotations (0.2.24)
  - add missing prefix property to auth backend (0.2.23)

oras/provider.py

@@ -274,6 +274,12 @@ def upload_blob(
         blob = os.path.abspath(blob)
         container = self.get_container(container)
 
+        if self.blob_exists(layer, container):
+            logger.debug(f'layer already exists: {layer["digest"]}')
+            response = requests.Response()
+            response.status_code = 200
+            return response
+
         # Chunked for large, otherwise POST and PUT
         # This is currently disabled unless the user asks for it, as
         # it doesn't seem to work for all registries
@@ -551,6 +557,21 @@ def put_upload(
             )
         return response
 
+    def blob_exists(
+        self, layer: oras.oci.Layer, container: oras.container.Container
+    ) -> bool:
+        """
+        Check if a layer already exists in the registry.
+
+        :param layer: the layer to check for existence
+        :type layer: oras.oci.Layer
+        :param container: the container to determine where to look for layer existence
+        :type container: oras.container.Container
+        """
+        blob_url = container.get_blob_url(layer["digest"])
+        response = self.do_request(f"{self.prefix}://{blob_url}", "HEAD")
+        return response.status_code == 200
+
     def _get_location(
         self, r: requests.Response, container: oras.container.Container
     ) -> str:
@@ -944,7 +965,7 @@ def do_request(
         headers: Optional[dict] = None,
         json: Optional[dict] = None,
         stream: bool = False,
-    ):
+    ) -> requests.Response:
         """
         Do a request. This is a wrapper around requests to handle retry auth.