Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ uploaded = bucket.upload_resumable(
"videos/automatic.mp4",
video,
content_type="video/mp4",
on_progress=lambda uploaded, total: print(f"{uploaded}/{total}"),
)
print(uploaded.name)
upload_session = bucket.create_upload_session(
Expand Down Expand Up @@ -110,7 +111,9 @@ part count, and expiration time for a resumable upload.
`upload_resumable()` accepts bytes or a binary file-like object, creates a
session, and uploads server-sized chunks. It streams seekable files directly;
non-seekable inputs are spooled to a temporary file with bounded reads. If a
part fails, it makes a best-effort abort and raises the original error.
part or progress callback fails, it makes a best-effort abort and raises the
original error. `on_progress` runs after each successful part with cumulative
uploaded bytes and the total size.
`upload_part()` returns immutable part metadata and can safely retry the same
part number to replace that part.
`locks.get()` returns immutable lock availability, expiry, and fencing-token
Expand Down
20 changes: 17 additions & 3 deletions src/volcano_sdk/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import base64
import binascii
import json
from collections.abc import Generator, Mapping, Sequence
from collections.abc import Callable, Generator, Mapping, Sequence
from contextlib import contextmanager, suppress
from dataclasses import dataclass
from datetime import datetime
Expand Down Expand Up @@ -564,6 +564,7 @@ def upload_resumable(
*,
content_type: str = "application/octet-stream",
part_size: int | None = None,
on_progress: Callable[[int, int], None] | None = None,
) -> StorageObject:
"""Upload bytes or a binary stream through a resumable session."""
path = _storage_path(path)
Expand All @@ -577,7 +578,13 @@ def upload_resumable(
)
upload_succeeded = False
try:
self._upload_session_parts(path, source, session)
self._upload_session_parts(
path,
source,
session,
total_size,
on_progress,
)
upload_succeeded = True
finally:
if not upload_succeeded:
Expand All @@ -589,14 +596,21 @@ def _upload_session_parts(
path: str,
source: BinaryIO,
session: UploadSession,
total_size: int,
on_progress: Callable[[int, int], None] | None,
) -> None:
uploaded = 0
for part_index in range(session.total_parts):
part = _read_upload_part(source, session.part_size)
self.upload_part(
path,
session_id=session.session_id,
part_number=part_index + 1,
data=_read_upload_part(source, session.part_size),
data=part,
)
uploaded += len(part)
if on_progress is not None:
on_progress(uploaded, total_size)

def _abort_failed_upload(self, path: str, session_id: str) -> None:
with suppress(Exception):
Expand Down
41 changes: 41 additions & 0 deletions tests/unit/test_facade.py
Original file line number Diff line number Diff line change
Expand Up @@ -816,6 +816,47 @@ def test_storage_uploads_bytes_with_server_selected_chunks() -> None:
assert object_.name == "videos/demo.mp4"


def test_storage_reports_progress_after_each_uploaded_part() -> 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")
progress: list[tuple[int, int]] = []

client.storage.from_("assets").upload_resumable(
"file.bin",
b"abcdefghij",
on_progress=lambda uploaded, total: progress.append((uploaded, total)),
)

assert progress == [(4, 10), (8, 10), (10, 10)]


def test_storage_aborts_when_a_progress_callback_fails() -> None:
transport = FakeTransport()
transport.upload_session_part_size = 4
transport.upload_session_total_parts = 2
client = VolcanoClient(anon_key="anon-key", _transport=transport)
client.auth.sign_in(email="user@example.com", password="secret")

def fail_progress(_uploaded: int, _total: int) -> None:
message = "progress failed"
raise RuntimeError(message)

with pytest.raises(RuntimeError, match="progress failed"):
client.storage.from_("assets").upload_resumable(
"file.bin",
b"abcdefgh",
on_progress=fail_progress,
)

assert [operation for operation, _ in transport.calls[-2:]] == [
"uploadPart",
"abortUploadSession",
]


def test_storage_aborts_after_part_failure_without_masking_the_error() -> None:
transport = FakeTransport()
transport.upload_session_part_size = 4
Expand Down
Loading