From 36e198ff153c925c2cef5d6f81bdfa2b30d74d74 Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Wed, 2 Sep 2026 12:19:40 -0400 Subject: [PATCH] feat(storage): orchestrate resumable uploads --- README.md | 9 ++++++ src/volcano_sdk/storage.py | 43 +++++++++++++++++++++++++ tests/unit/test_facade.py | 66 ++++++++++++++++++++++++++++++++++++-- 3 files changed, 116 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 251195b7..ee9726b5 100644 --- a/README.md +++ b/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 diff --git a/src/volcano_sdk/storage.py b/src/volcano_sdk/storage.py index 5efe5977..1292421c 100644 --- a/src/volcano_sdk/storage.py +++ b/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 = "", diff --git a/tests/unit/test_facade.py b/tests/unit/test_facade.py index 4cc16d36..5870e8c0 100644 --- a/tests/unit/test_facade.py +++ b/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,8 +82,8 @@ 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", }, ) @@ -86,6 +91,8 @@ def create_upload_session(self, **kwargs: Any) -> FakeResponse: 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,