Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
78d500d
feat(storage): support full_object_checksum in AsyncAppendableObjectW…
chandra-siri Jul 8, 2026
b231244
test(storage): add unit tests for AsyncAppendableObjectWriter checksu…
chandra-siri Jul 8, 2026
5b518be
test(storage): add integration tests for AsyncAppendableObjectWriter …
chandra-siri Jul 8, 2026
27933da
style: run ruff format on test_async_appendable_object_writer.py
chandra-siri Jul 8, 2026
22ffa69
refactor(storage): conditionally attach object_checksums to finalizat…
chandra-siri Jul 8, 2026
7a58986
refactor(storage): replace warning with comment when checksum is None
chandra-siri Jul 8, 2026
9e02476
Revert "refactor(storage): replace warning with comment when checksum…
chandra-siri Jul 8, 2026
063142b
test(storage): assert exceptions.InvalidArgument on checksum mismatch…
chandra-siri Jul 8, 2026
7dd7cd8
feat(storage): validate type and range of full_object_checksum
chandra-siri Jul 8, 2026
0a73153
fix(storage): close stream and reset local state in finalize on error
chandra-siri Jul 8, 2026
f7b8e4c
feat(storage): allow option to disable checksums for faster async app…
chandra-siri Jul 8, 2026
277eb1a
refactor(storage): remove warning log when full_object_checksum is No…
chandra-siri Jul 9, 2026
9b13058
test(storage): change integration tests object name prefix to appenda…
chandra-siri Jul 9, 2026
c7096b7
merge branch 'feat/appendable-checksum-validation' into feat/appendab…
chandra-siri Jul 9, 2026
94d04fa
refactor(storage): simplify checksum checking and object construction…
chandra-siri Jul 9, 2026
6cb0cd2
chore(storage) resolve merge conflicts
chandra-siri Jul 9, 2026
6f2dad4
chore(storage): remove accidentally committed local/untracked files f…
chandra-siri Jul 9, 2026
d8187ba
fix(storage): wrap finalize request send within try...finally block
chandra-siri Jul 9, 2026
410fa06
fix(storage): explicitly reject boolean types for full_object_checksu…
chandra-siri Jul 9, 2026
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
Original file line number Diff line number Diff line change
Expand Up @@ -386,6 +386,7 @@ async def append(
data: bytes,
retry_policy: Optional[AsyncRetry] = None,
metadata: Optional[List[Tuple[str, str]]] = None,
enable_checksum: bool = True,
) -> None:
"""Appends data to the Appendable object with automatic retries.

Expand All @@ -406,6 +407,9 @@ async def append(
:type metadata: List[Tuple[str, str]]
:param metadata: (Optional) The metadata to be sent with the request.

:type enable_checksum: bool
:param enable_checksum: (Optional) If True, calculates and checks checksums for each chunk. Defaults to True.

:raises ValueError: If the stream is not open.
"""
if not self._is_stream_open:
Expand Down Expand Up @@ -487,7 +491,12 @@ async def generator():
return generator()

# State initialization
write_state = _WriteState(_MAX_CHUNK_SIZE_BYTES, buffer, self.flush_interval)
write_state = _WriteState(
_MAX_CHUNK_SIZE_BYTES,
buffer,
self.flush_interval,
enable_checksum=enable_checksum,
)
write_state.write_handle = self.write_handle
write_state.persisted_size = self.persisted_size
# offset is set during `open()` call.
Expand Down Expand Up @@ -641,22 +650,32 @@ async def finalize(
if not self._is_stream_open:
raise ValueError("Stream is not open. Call open() before finalize().")

finalize_req = _storage_v2.BidiWriteObjectRequest(finish_write=True)

if full_object_checksum is not None:
finalize_req.object_checksums = _storage_v2.ObjectChecksums(
crc32c=full_object_checksum
if full_object_checksum is None:
finalize_req = _storage_v2.BidiWriteObjectRequest(finish_write=True)
elif isinstance(full_object_checksum, bool) or not isinstance(
full_object_checksum, int
):
raise TypeError("full_object_checksum must be an integer.")
elif not (0 <= full_object_checksum <= 0xFFFFFFFF):
raise ValueError("full_object_checksum must be a 32-bit unsigned integer.")
else:
finalize_req = _storage_v2.BidiWriteObjectRequest(
finish_write=True,
object_checksums=_storage_v2.ObjectChecksums(
crc32c=full_object_checksum
),
)

await self.write_obj_stream.send(finalize_req)
response = await self.write_obj_stream.recv()
self.object_resource = response.resource
self.persisted_size = self.object_resource.size
await self.write_obj_stream.close()

self._is_stream_open = False
self.offset = None
return self.object_resource
try:
await self.write_obj_stream.send(finalize_req)
response = await self.write_obj_stream.recv()
self.object_resource = response.resource
self.persisted_size = self.object_resource.size
return self.object_resource
finally:
await self.write_obj_stream.close()
self._is_stream_open = False
self.offset = None

@property
def is_stream_open(self) -> bool:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -407,6 +407,12 @@ async def download_ranges(
:type retry_policy: :class:`~google.api_core.retry_async.AsyncRetry`
:param retry_policy: (Optional) The retry policy to use for the operation.

:type metadata: List[Tuple[str, str]]
:param metadata: (Optional) The metadata to be sent with the request.

:type enable_checksum: bool
:param enable_checksum: (Optional) If True, checksums are verified for downloaded data. Defaults to True.

Comment thread
chandra-siri marked this conversation as resolved.
:raises ValueError: if the underlying bidi-GRPC stream is not open.
:raises ValueError: if the length of read_ranges is more than 1000.
:raises DataCorruption: if a checksum mismatch is detected while reading data.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ def __init__(
chunk_size: int,
user_buffer: IO[bytes],
flush_interval: int,
enable_checksum: bool = True,
):
self.chunk_size = chunk_size
self.user_buffer = user_buffer
Expand All @@ -59,6 +60,7 @@ def __init__(
self.write_handle: Union[bytes, storage_type.BidiWriteHandle, None] = None
self.routing_token: Optional[str] = None
self.is_finalized: bool = False
self.enable_checksum: bool = enable_checksum


class _WriteResumptionStrategy(_BaseResumptionStrategy):
Expand All @@ -84,7 +86,8 @@ def generate_requests(
break

checksummed_data = storage_type.ChecksummedData(content=chunk)
checksummed_data.crc32c = google_crc32c.value(chunk)
if write_state.enable_checksum:
checksummed_data.crc32c = google_crc32c.value(chunk)

request = storage_type.BidiWriteObjectRequest(
write_offset=write_state.bytes_sent,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,22 @@ def test_generate_requests_checksum_verification(self, strategy):
expected_int = google_crc32c.value(chunk_data)
assert requests[0].checksummed_data.crc32c == expected_int

def test_generate_requests_checksum_disabled(self, strategy):
"""Verify CRC32C is not calculated if enable_checksum is False."""
chunk_data = b"test_data"
mock_buffer = io.BytesIO(chunk_data)
write_state = _WriteState(
chunk_size=10,
user_buffer=mock_buffer,
flush_interval=10,
enable_checksum=False,
)
state = {"write_state": write_state}

requests = strategy.generate_requests(state)

assert not requests[0].checksummed_data.crc32c

def test_generate_requests_flush_logic_exact_interval(self, strategy):
"""Verify the flush bit is set exactly when the interval is reached."""
mock_buffer = io.BytesIO(b"A" * 12)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -299,6 +299,29 @@ async def test_append_data_less_than_flush_interval(self, mock_appendable_writer
assert writer.offset == data_len
assert writer.bytes_appended_since_last_flush == data_len

@pytest.mark.asyncio
async def test_append_with_checksum_disabled(self, mock_appendable_writer):
"""Verify append propagates enable_checksum=False to the _WriteState."""
writer = self._make_one(mock_appendable_writer["mock_client"])
writer._is_stream_open = True
writer.persisted_size = 0
writer.write_obj_stream = mock_appendable_writer["mock_stream"]
writer.write_obj_stream.send = AsyncMock()

with mock.patch(
"google.cloud.storage.asyncio.async_appendable_object_writer._BidiStreamRetryManager"
) as MockManager:
mock_execute = AsyncMock()
MockManager.return_value.execute = mock_execute

await writer.append(DATA_LESS_THAN_FLUSH_INTERVAL, enable_checksum=False)

# Check that execute was called with a state dictionary containing write_state having enable_checksum=False
mock_execute.assert_called_once()
state_arg = mock_execute.call_args[0][0]
assert "write_state" in state_arg
assert state_arg["write_state"].enable_checksum is False

@pytest.mark.parametrize(
"data_len",
[
Expand Down Expand Up @@ -552,3 +575,53 @@ async def test_close_with_checksum_without_finalize_raises(
match="full_object_checksum can only be provided when finalize_on_close is True",
):
await writer.close(finalize_on_close=False, full_object_checksum=checksum)

@pytest.mark.asyncio
async def test_finalize_invalid_checksum_type(self, mock_appendable_writer):
writer = self._make_one(mock_appendable_writer["mock_client"])
writer._is_stream_open = True
writer.write_obj_stream = mock_appendable_writer["mock_stream"]

with pytest.raises(TypeError, match="full_object_checksum must be an integer"):
await writer.finalize(full_object_checksum="not-an-int")
Comment thread
chandra-siri marked this conversation as resolved.

with pytest.raises(TypeError, match="full_object_checksum must be an integer"):
await writer.finalize(full_object_checksum=True)

@pytest.mark.asyncio
async def test_finalize_invalid_checksum_range(self, mock_appendable_writer):
writer = self._make_one(mock_appendable_writer["mock_client"])
writer._is_stream_open = True
writer.write_obj_stream = mock_appendable_writer["mock_stream"]

# negative
with pytest.raises(
ValueError, match="full_object_checksum must be a 32-bit unsigned integer"
):
await writer.finalize(full_object_checksum=-1)

# overflow
with pytest.raises(
ValueError, match="full_object_checksum must be a 32-bit unsigned integer"
):
await writer.finalize(full_object_checksum=0x100000000)

@pytest.mark.asyncio
async def test_finalize_mismatch_closes_stream(self, mock_appendable_writer):
writer = self._make_one(mock_appendable_writer["mock_client"])
writer._is_stream_open = True
writer.write_obj_stream = mock_appendable_writer["mock_stream"]

# Mock recv to raise an exception (like server rejecting checksum mismatch)
from google.api_core.exceptions import InvalidArgument

mock_appendable_writer["mock_stream"].recv.side_effect = InvalidArgument(
Comment thread
chandra-siri marked this conversation as resolved.
"checksum mismatch"
)

with pytest.raises(InvalidArgument):
await writer.finalize(full_object_checksum=12345)

# Assert stream was closed and local state reset despite exception
mock_appendable_writer["mock_stream"].close.assert_awaited()
assert not writer._is_stream_open
Loading