feat(storage): orchestrate resumable uploads

README.md

@@ -46,6 +46,12 @@ downloaded = bucket.download("a.txt")
 assert downloaded == b"hello"
 first_kibibyte = bucket.download("archive.bin", byte_range="bytes=0-1023")
 video = b"demo video"
+uploaded = bucket.upload_resumable(
+    "videos/automatic.mp4",
+    video,
+    content_type="video/mp4",
+)
+print(uploaded.name)
 upload_session = bucket.create_upload_session(
     "videos/demo.mp4",
     total_size=len(video),
@@ -97,6 +103,9 @@ object; `public_url` is set only when the object is public.
 Pass an HTTP byte range to download only part of an object.
 `create_upload_session()` returns the immutable server-selected part size,
 part count, and expiration time for a resumable upload.
+`upload_resumable()` creates a session, chunks the bytes using the
+server-selected part size, and completes the upload. If a part fails, it makes
+a best-effort abort and raises the original error.
 `upload_part()` returns immutable part metadata and can safely retry the same
 part number to replace that part.
 `get_upload_session()` returns immutable progress and uploaded-part metadata for

src/volcano_sdk/storage.py

@@ -6,6 +6,7 @@
 import binascii
 import json
 from collections.abc import Mapping, Sequence
+from contextlib import suppress
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Protocol, cast
@@ -20,6 +21,7 @@
     invoke,
     response_payload,
 )
+from .errors import VolcanoError
 from .models import (
     JSONValue,
     StorageObject,
@@ -493,6 +495,47 @@ def abort_upload_session(
         )
         response_payload(response, 200)
 
+    def upload_resumable(
+        self,
+        path: str,
+        data: bytes,
+        *,
+        content_type: str = "application/octet-stream",
+        part_size: int | None = None,
+    ) -> StorageObject:
+        """Upload bytes through a server-managed resumable session."""
+        session = self.create_upload_session(
+            path,
+            total_size=len(data),
+            content_type=content_type,
+            part_size=part_size,
+        )
+        try:
+            self._upload_session_parts(path, data, session)
+        except VolcanoError:
+            self._abort_failed_upload(path, session.session_id)
+            raise
+        return self.complete_upload_session(path, session_id=session.session_id)
+
+    def _upload_session_parts(
+        self,
+        path: str,
+        data: bytes,
+        session: UploadSession,
+    ) -> None:
+        for part_index in range(session.total_parts):
+            offset = part_index * session.part_size
+            self.upload_part(
+                path,
+                session_id=session.session_id,
+                part_number=part_index + 1,
+                data=data[offset : offset + session.part_size],
+            )
+
+    def _abort_failed_upload(self, path: str, session_id: str) -> None:
+        with suppress(VolcanoError):
+            self.abort_upload_session(path, session_id=session_id)
+
     def list(
         self,
         prefix: str = "",

tests/unit/test_facade.py

@@ -10,6 +10,7 @@
 
 from volcano_sdk import (
     LockLease,
+    ServerError,
     Session,
     StorageObject,
     StoragePage,
@@ -34,6 +35,10 @@ def __init__(self) -> None:
         self.list_cursor = "cursor-2"
         self.range_download_status = 206
         self.include_upload_session_parts = True
+        self.upload_session_part_size = 8_388_608
+        self.upload_session_total_parts = 3
+        self.fail_upload_part_number: int | None = None
+        self.fail_abort_upload = False
 
     def auth_signin(self, **kwargs: Any) -> FakeResponse:
         self.calls.append(("authSignin", kwargs))
@@ -77,15 +82,17 @@ def create_upload_session(self, **kwargs: Any) -> FakeResponse:
             201,
             {
                 "session_id": "session-123",
-                "part_size": 8_388_608,
-                "total_parts": 3,
+                "part_size": self.upload_session_part_size,
+                "total_parts": self.upload_session_total_parts,
                 "expires_at": "2026-09-09T12:00:00Z",
             },
         )
 
     def upload_part(self, **kwargs: Any) -> FakeResponse:
         self.calls.append(("uploadPart", kwargs))
         request = kwargs["request"]
+        if request.part_number == self.fail_upload_part_number:
+            return FakeResponse(500, {"error": "part upload failed"})
         return FakeResponse(
             200,
             {
@@ -141,6 +148,8 @@ def get_upload_session(self, **kwargs: Any) -> FakeResponse:
 
     def abort_upload_session(self, **kwargs: Any) -> FakeResponse:
         self.calls.append(("abortUploadSession", kwargs))
+        if self.fail_abort_upload:
+            return FakeResponse(500, {"error": "abort failed"})
         return FakeResponse(200, {"message": "upload session aborted"})
 
     def list_storage_objects(self, **kwargs: Any) -> FakeResponse:
@@ -585,6 +594,59 @@ def test_storage_aborts_an_upload_session() -> None:
     )
 
 
+def test_storage_uploads_bytes_with_server_selected_chunks() -> None:
+    transport = FakeTransport()
+    transport.upload_session_part_size = 4
+    transport.upload_session_total_parts = 3
+    client = VolcanoClient(anon_key="anon-key", _transport=transport)
+    client.auth.sign_in(email="user@example.com", password="secret")
+
+    object_ = client.storage.from_("assets").upload_resumable(
+        "videos/demo.mp4",
+        b"abcdefghij",
+        content_type="video/mp4",
+        part_size=6,
+    )
+
+    storage_calls = transport.calls[1:]
+    assert [operation for operation, _ in storage_calls] == [
+        "createUploadSession",
+        "uploadPart",
+        "uploadPart",
+        "uploadPart",
+        "completeUploadSession",
+    ]
+    create_request = storage_calls[0][1]["request"]
+    assert (create_request.total_size, create_request.part_size) == (10, 6)
+    assert [call[1]["request"].data for call in storage_calls[1:4]] == [
+        b"abcd",
+        b"efgh",
+        b"ij",
+    ]
+    assert object_.name == "videos/demo.mp4"
+
+
+def test_storage_aborts_after_part_failure_without_masking_the_error() -> None:
+    transport = FakeTransport()
+    transport.upload_session_part_size = 4
+    transport.upload_session_total_parts = 2
+    transport.fail_upload_part_number = 2
+    transport.fail_abort_upload = True
+    client = VolcanoClient(anon_key="anon-key", _transport=transport)
+    client.auth.sign_in(email="user@example.com", password="secret")
+
+    with pytest.raises(ServerError, match="part upload failed") as raised:
+        client.storage.from_("assets").upload_resumable("file.bin", b"abcdefgh")
+
+    assert getattr(raised.value, "status", None) == 500
+    assert [operation for operation, _ in transport.calls[1:]] == [
+        "createUploadSession",
+        "uploadPart",
+        "uploadPart",
+        "abortUploadSession",
+    ]
+
+
 @pytest.mark.parametrize("invalid_paths", [[], [""], b"abc"])
 def test_storage_remove_rejects_invalid_paths_before_transport(
     invalid_paths: Any,