[EDPOPS-144] Add multipart upload support (#498)

* Set py 3.8 as required version

* [EDPOPS-144] Add multipart upload enhancements

* linting

* linting

* linting

---------

Co-authored-by: Tatjana Damnjanović <tatjana@solvebio.com>

Author: damnjanovictanja

solvebio/resource/object.py

@@ -434,8 +434,12 @@ def upload_file(cls, local_path, remote_path, vault_full_path, **kwargs):
         # Get vault
         vault = Vault.get_by_full_path(vault_full_path, client=_client)
 
-        # Get MD5
-        local_md5, _ = md5sum(local_path, multipart_threshold=None)
+        # Get MD5 and check if multipart upload is needed
+        multipart_threshold = kwargs.get(
+            "multipart_threshold", 64 * 1024 * 1024
+        )  # 64MB default
+        local_md5, _ = md5sum(local_path, multipart_threshold=multipart_threshold)
+
         # Get a mimetype of file
         mime_tuple = mimetypes.guess_type(local_path)
         # If a file is compressed get a compression type, otherwise a file type
@@ -509,6 +513,27 @@ def upload_file(cls, local_path, remote_path, vault_full_path, **kwargs):
                                                            obj.path))
         print('Notice: Upload initialized')
 
+        # Check if multipart upload is needed
+        if hasattr(obj, "is_multipart") and obj.is_multipart:
+            return cls._upload_multipart(obj, local_path, local_md5, **kwargs)
+        else:
+            return cls._upload_single_file(obj, local_path, **kwargs)
+
+    @classmethod
+    def _upload_single_file(cls, obj, local_path, **kwargs):
+        """Handle single-part upload for smaller files"""
+        import mimetypes
+
+        # Get a mimetype of file
+        mime_tuple = mimetypes.guess_type(local_path)
+        # If a file is compressed get a compression type, otherwise a file type
+        mimetype = mime_tuple[1] if mime_tuple[1] else mime_tuple[0]
+        # Get file size
+        size = os.path.getsize(local_path)
+
+        # Get MD5 for single part upload
+        local_md5, _ = md5sum(local_path, multipart_threshold=None)
+
         upload_url = obj.upload_url
 
         headers = {
@@ -563,6 +588,109 @@ def upload_file(cls, local_path, remote_path, vault_full_path, **kwargs):
 
         return obj
 
+    @classmethod
+    def _upload_multipart(cls, obj, local_path, local_md5, **kwargs):
+        """Handle multipart upload for larger files"""
+        _client = kwargs.get("client") or cls._client or client
+        print(f"Notice: Upload ID {obj.upload_id}")
+        try:
+            # Get presigned URLs from the object
+            presigned_urls = obj.presigned_urls
+
+            print(
+                "Notice: Starting multipart upload with {} parts...".format(
+                    len(presigned_urls)
+                )
+            )
+
+            # Step 2: Upload each part using presigned URLs
+            parts = []
+            with open(local_path, "rb") as f:
+                for part_info in presigned_urls:
+                    part_number = part_info.part_number
+                    start_byte = part_info.start_byte
+                    end_byte = part_info.end_byte
+                    part_size = part_info.size
+                    upload_url = part_info.upload_url
+
+                    print(
+                        "Notice: Uploading part {}/{}... (bytes {}-{})".format(
+                            part_number, len(presigned_urls), start_byte, end_byte
+                        )
+                    )
+
+                    # Seek to start position and read the exact part size
+                    f.seek(start_byte)
+                    chunk_data = f.read(part_size)
+                    if not chunk_data:
+                        break
+
+                    # Upload part with retry logic
+                    session = requests.Session()
+                    retry = Retry(
+                        total=3,
+                        backoff_factor=2,
+                        status_forcelist=(500, 502, 503, 504),
+                        allowed_methods=["PUT"],
+                    )
+                    session.mount(
+                        "https://", requests.adapters.HTTPAdapter(max_retries=retry)
+                    )
+
+                    headers = {
+                        "Content-Length": str(len(chunk_data)),
+                    }
+
+                    upload_resp = session.put(
+                        upload_url, data=chunk_data, headers=headers
+                    )
+
+                    if upload_resp.status_code != 200:
+                        raise FileUploadError(
+                            "Failed to upload part {}: {}".format(
+                                part_number, upload_resp.content
+                            )
+                        )
+
+                    # Get ETag from response
+                    etag = upload_resp.headers.get("ETag", "").strip('"')
+                    parts.append({"part_number": part_number, "etag": etag})
+
+            # Step 3: Complete multipart upload
+            print("Notice: Completing multipart upload....")
+            complete_data = {
+                "upload_id": obj.upload_id,
+                "physical_object_id": obj.upload_key,
+                "parts": parts,
+            }
+
+            print(f"Notice: {complete_data}")
+
+            complete_resp = _client.post("/v2/complete_multi_part", complete_data)
+
+            if "message" in complete_resp:
+                print(
+                    "Notice: Successfully uploaded {0} to {1} with multipart upload.".format(
+                        local_path, obj.path
+                    )
+                )
+                return obj
+            else:
+                raise Exception(complete_resp)
+
+        except Exception as e:
+            # Clean up failed upload - best effort cleanup
+            try:
+                _client.delete(
+                    obj.instance_url() + "/multipart-upload",
+                    {},
+                )
+            except Exception:
+                pass  # Best effort cleanup
+
+            obj.delete(force=True)
+            raise FileUploadError("Multipart upload failed: {}".format(str(e)))
+
     def _object_list_helper(self, **params):
         """Helper method to get objects within"""