feat: compute chunk wise checksum for bidi_writes (#1675)

feat: compute chunk wise checksum for bidi_writes and send it via
BidiWriteObjectRequest

As a part of this change, also did a small refactoring 
  *  Moved the precondition check to __utils.py_ file

Author: chandra-siri

google/cloud/storage/_experimental/asyncio/_utils.py

@@ -0,0 +1,34 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import google_crc32c
+
+from google.api_core import exceptions
+
+def raise_if_no_fast_crc32c():
+    """Check if the C-accelerated version of google-crc32c is available.
+
+    If not, raise an error to prevent silent performance degradation.
+
+    raises google.api_core.exceptions.FailedPrecondition: If the C extension is not available.
+    returns: True if the C extension is available.
+    rtype: bool
+    
+    """
+    if google_crc32c.implementation != "c":
+        raise exceptions.FailedPrecondition(
+            "The google-crc32c package is not installed with C support. "
+            "C extension is required for faster data integrity checks."
+            "For more information, see https://github.com/googleapis/python-crc32c."
+        )

google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py

@@ -22,6 +22,10 @@
 
 """
 from typing import Optional, Union
+
+from google_crc32c import Checksum
+
+from ._utils import raise_if_no_fast_crc32c
 from google.cloud import _storage_v2
 from google.cloud.storage._experimental.asyncio.async_grpc_client import (
     AsyncGrpcClient,
@@ -100,6 +104,7 @@ def __init__(
         :param write_handle: (Optional) An existing handle for writing the object.
                             If provided, opening the bidi-gRPC connection will be faster.
         """
+        raise_if_no_fast_crc32c()
         self.client = client
         self.bucket_name = bucket_name
         self.object_name = object_name
@@ -191,11 +196,13 @@ async def append(self, data: bytes) -> None:
         bytes_to_flush = 0
         while start_idx < total_bytes:
             end_idx = min(start_idx + _MAX_CHUNK_SIZE_BYTES, total_bytes)
+            data_chunk = data[start_idx:end_idx]
             await self.write_obj_stream.send(
                 _storage_v2.BidiWriteObjectRequest(
                     write_offset=self.offset,
                     checksummed_data=_storage_v2.ChecksummedData(
-                        content=data[start_idx:end_idx]
+                        content=data_chunk,
+                        crc32c=int.from_bytes(Checksum(data_chunk).digest(), "big"),
                     ),
                 )
             )

google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py

@@ -14,12 +14,11 @@
 
 from __future__ import annotations
 import asyncio
-import google_crc32c
-from google.api_core import exceptions
-from google_crc32c import Checksum
-
 from typing import List, Optional, Tuple
 
+from google_crc32c import Checksum
+
+from ._utils import raise_if_no_fast_crc32c
 from google.cloud.storage._experimental.asyncio.async_read_object_stream import (
     _AsyncReadObjectStream,
 )
@@ -160,14 +159,7 @@ def __init__(
         :param read_handle: (Optional) An existing read handle.
         """
 
-        # Verify that the fast, C-accelerated version of crc32c is available.
-        # If not, raise an error to prevent silent performance degradation.
-        if google_crc32c.implementation != "c":
-            raise exceptions.NotFound(
-                "The google-crc32c package is not installed with C support. "
-                "Bidi reads require the C extension for data integrity checks."
-                "For more information, see https://github.com/googleapis/python-crc32c."
-            )
+        raise_if_no_fast_crc32c()
 
         self.client = client
         self.bucket_name = bucket_name

tests/unit/asyncio/test_async_appendable_object_writer.py

@@ -15,6 +15,9 @@
 import pytest
 from unittest import mock
 
+from google_crc32c import Checksum
+
+from google.api_core import exceptions
 from google.cloud.storage._experimental.asyncio.async_appendable_object_writer import (
     AsyncAppendableObjectWriter,
 )
@@ -85,6 +88,23 @@ def test_init_with_optional_args(mock_write_object_stream, mock_client):
     )
 
 
+@mock.patch("google.cloud.storage._experimental.asyncio._utils.google_crc32c")
+@mock.patch(
+    "google.cloud.storage._experimental.asyncio.async_grpc_client.AsyncGrpcClient.grpc_client"
+)
+def test_init_raises_if_crc32c_c_extension_is_missing(
+    mock_grpc_client, mock_google_crc32c
+):
+    mock_google_crc32c.implementation = "python"
+
+    with pytest.raises(exceptions.FailedPrecondition) as exc_info:
+        AsyncAppendableObjectWriter(mock_grpc_client, "bucket", "object")
+
+    assert "The google-crc32c package is not installed with C support" in str(
+        exc_info.value
+    )
+
+
 @pytest.mark.asyncio
 @mock.patch(
     "google.cloud.storage._experimental.asyncio.async_appendable_object_writer._AsyncWriteObjectStream"
@@ -434,10 +454,15 @@ async def test_append_sends_data_in_chunks(mock_write_object_stream, mock_client
     # First chunk
     assert first_call[0][0].write_offset == 100
     assert len(first_call[0][0].checksummed_data.content) == _MAX_CHUNK_SIZE_BYTES
-
+    assert first_call[0][0].checksummed_data.crc32c == int.from_bytes(
+        Checksum(data[:_MAX_CHUNK_SIZE_BYTES]).digest(), byteorder="big"
+    )
     # Second chunk
     assert second_call[0][0].write_offset == 100 + _MAX_CHUNK_SIZE_BYTES
     assert len(second_call[0][0].checksummed_data.content) == 1
+    assert second_call[0][0].checksummed_data.crc32c == int.from_bytes(
+        Checksum(data[_MAX_CHUNK_SIZE_BYTES:]).digest(), byteorder="big"
+    )
 
     assert writer.offset == 100 + len(data)
     writer.simple_flush.assert_not_awaited()

tests/unit/asyncio/test_async_multi_range_downloader.py

@@ -349,9 +349,7 @@ async def test_downloading_without_opening_should_throw_error(
         assert str(exc.value) == "Underlying bidi-gRPC stream is not open"
         assert not mrd.is_stream_open
 
-    @mock.patch(
-        "google.cloud.storage._experimental.asyncio.async_multi_range_downloader.google_crc32c"
-    )
+    @mock.patch("google.cloud.storage._experimental.asyncio._utils.google_crc32c")
     @mock.patch(
         "google.cloud.storage._experimental.asyncio.async_grpc_client.AsyncGrpcClient.grpc_client"
     )
@@ -360,7 +358,7 @@ def test_init_raises_if_crc32c_c_extension_is_missing(
     ):
         mock_google_crc32c.implementation = "python"
 
-        with pytest.raises(exceptions.NotFound) as exc_info:
+        with pytest.raises(exceptions.FailedPrecondition) as exc_info:
             AsyncMultiRangeDownloader(mock_grpc_client, "bucket", "object")
 
         assert "The google-crc32c package is not installed with C support" in str(