Merge branch 'main' into fix_hide_sensitive_query_keys

Author: rajeevpodar

CHANGELOG.md

@@ -4,6 +4,12 @@
 
 [1]: https://pypi.org/project/google-cloud-storage/#history
 
+## [3.4.1](https://github.com/googleapis/python-storage/compare/v3.4.0...v3.5.0) (2025-10-08)
+
+### Bug Fixes
+
+* Fixes [#1561](https://github.com/googleapis/python-storage/issues/1561) by adding an option to specify the entire object checksum for resumable uploads via the `upload_from_string`, `upload_from_file`, and `upload_from_filename` methods ([acb918e](https://github.com/googleapis/python-storage/commit/acb918e20f7092e13d72fc63fe4ae2560bfecd40))
+
 ## [3.4.0](https://github.com/googleapis/python-storage/compare/v3.3.1...v3.4.0) (2025-09-15)
 
 

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

@@ -13,8 +13,11 @@
 # limitations under the License.
 
 from __future__ import annotations
+import google_crc32c
+from google.api_core import exceptions
+from google_crc32c import Checksum
 
-from typing import Any, List, Optional, Tuple
+from typing import List, Optional, Tuple
 
 from google.cloud.storage._experimental.asyncio.async_read_object_stream import (
     _AsyncReadObjectStream,
@@ -24,6 +27,39 @@
 )
 
 from io import BytesIO
+from google.cloud import _storage_v2
+from google.cloud.storage.exceptions import DataCorruption
+
+
+_MAX_READ_RANGES_PER_BIDI_READ_REQUEST = 100
+
+
+class Result:
+    """An instance of this class will be populated and retured for each
+    `read_range` provided to ``download_ranges`` method.
+
+    """
+
+    def __init__(self, bytes_requested: int):
+        # only while instantiation, should not be edited later.
+        # hence there's no setter, only getter is provided.
+        self._bytes_requested: int = bytes_requested
+        self._bytes_written: int = 0
+
+    @property
+    def bytes_requested(self) -> int:
+        return self._bytes_requested
+
+    @property
+    def bytes_written(self) -> int:
+        return self._bytes_written
+
+    @bytes_written.setter
+    def bytes_written(self, value: int):
+        self._bytes_written = value
+
+    def __repr__(self):
+        return f"bytes_requested: {self._bytes_requested}, bytes_written: {self._bytes_written}"
 
 
 class AsyncMultiRangeDownloader:
@@ -38,21 +74,23 @@ class AsyncMultiRangeDownloader:
         mrd = await AsyncMultiRangeDownloader.create_mrd(
             client, bucket_name="chandrasiri-rs", object_name="test_open9"
         )
-        my_buff1 = BytesIO()
+        my_buff1 = open('my_fav_file.txt', 'wb')
         my_buff2 = BytesIO()
         my_buff3 = BytesIO()
-        my_buff4 = BytesIO()
-        buffers = [my_buff1, my_buff2, my_buff3, my_buff4]
-        await mrd.download_ranges(
+        my_buff4 = any_object_which_provides_BytesIO_like_interface()
+        results_arr = await mrd.download_ranges(
             [
+                # (start_byte, bytes_to_read, writeable_buffer)
                 (0, 100, my_buff1),
-                (100, 200, my_buff2),
-                (200, 300, my_buff3),
-                (300, 400, my_buff4),
+                (100, 20, my_buff2),
+                (200, 123, my_buff3),
+                (300, 789, my_buff4),
             ]
         )
-        for buff in buffers:
-            print("downloaded bytes", buff.getbuffer().nbytes)
+
+        for result in results_arr:
+            print("downloaded bytes", result)
+
 
     """
 
@@ -119,12 +157,23 @@ def __init__(
         :type read_handle: bytes
         :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."
+            )
+
         self.client = client
         self.bucket_name = bucket_name
         self.object_name = object_name
         self.generation_number = generation_number
         self.read_handle = read_handle
-        self.read_obj_str: _AsyncReadObjectStream = None
+        self.read_obj_str: Optional[_AsyncReadObjectStream] = None
+        self._is_stream_open: bool = False
 
     async def open(self) -> None:
         """Opens the bidi-gRPC connection to read from the object.
@@ -135,31 +184,118 @@ async def open(self) -> None:
         "Opening" constitutes fetching object metadata such as generation number
         and read handle and sets them as attributes if not already set.
         """
-        self.read_obj_str = _AsyncReadObjectStream(
-            client=self.client,
-            bucket_name=self.bucket_name,
-            object_name=self.object_name,
-            generation_number=self.generation_number,
-            read_handle=self.read_handle,
-        )
+        if self._is_stream_open:
+            raise ValueError("Underlying bidi-gRPC stream is already open")
+
+        if self.read_obj_str is None:
+            self.read_obj_str = _AsyncReadObjectStream(
+                client=self.client,
+                bucket_name=self.bucket_name,
+                object_name=self.object_name,
+                generation_number=self.generation_number,
+                read_handle=self.read_handle,
+            )
         await self.read_obj_str.open()
+        self._is_stream_open = True
         if self.generation_number is None:
             self.generation_number = self.read_obj_str.generation_number
         self.read_handle = self.read_obj_str.read_handle
         return
 
-    async def download_ranges(self, read_ranges: List[Tuple[int, int, BytesIO]]) -> Any:
+    async def download_ranges(
+        self, read_ranges: List[Tuple[int, int, BytesIO]]
+    ) -> List[Result]:
         """Downloads multiple byte ranges from the object into the buffers
         provided by user.
 
         :type read_ranges: List[Tuple[int, int, "BytesIO"]]
         :param read_ranges: A list of tuples, where each tuple represents a
-            byte range (start_byte, end_byte, buffer) to download. Buffer has to
-            be provided by the user, and user has to make sure appropriate
+            byte range (start_byte, bytes_to_read, writeable_buffer). Buffer has
+            to be provided by the user, and user has to make sure appropriate
             memory is available in the application to avoid out-of-memory crash.
 
+        :rtype: List[:class:`~google.cloud.storage._experimental.asyncio.async_multi_range_downloader.Result`]
+        :returns: A list of ``Result`` objects, where each object corresponds
+                  to a requested range.
 
-        Raises:
-            NotImplementedError: This method is not yet implemented.
         """
-        raise NotImplementedError("TODO")
+
+        if len(read_ranges) > 1000:
+            raise ValueError(
+                "Invalid input - length of read_ranges cannot be more than 1000"
+            )
+
+        if not self._is_stream_open:
+            raise ValueError("Underlying bidi-gRPC stream is not open")
+
+        read_id_to_writable_buffer_dict = {}
+        results = []
+        for i in range(0, len(read_ranges), _MAX_READ_RANGES_PER_BIDI_READ_REQUEST):
+            read_ranges_segment = read_ranges[
+                i : i + _MAX_READ_RANGES_PER_BIDI_READ_REQUEST
+            ]
+
+            read_ranges_for_bidi_req = []
+            for j, read_range in enumerate(read_ranges_segment):
+                read_id = i + j
+                read_id_to_writable_buffer_dict[read_id] = read_range[2]
+                bytes_requested = read_range[1]
+                results.append(Result(bytes_requested))
+                read_ranges_for_bidi_req.append(
+                    _storage_v2.ReadRange(
+                        read_offset=read_range[0],
+                        read_length=bytes_requested,
+                        read_id=read_id,
+                    )
+                )
+            await self.read_obj_str.send(
+                _storage_v2.BidiReadObjectRequest(read_ranges=read_ranges_for_bidi_req)
+            )
+
+        while len(read_id_to_writable_buffer_dict) > 0:
+            response = await self.read_obj_str.recv()
+
+            if response is None:
+                raise Exception("None response received, something went wrong.")
+
+            for object_data_range in response.object_data_ranges:
+                if object_data_range.read_range is None:
+                    raise Exception("Invalid response, read_range is None")
+
+                checksummed_data = object_data_range.checksummed_data
+                data = checksummed_data.content
+                server_checksum = checksummed_data.crc32c
+
+                client_crc32c = Checksum(data).digest()
+                client_checksum = int.from_bytes(client_crc32c, "big")
+
+                if server_checksum != client_checksum:
+                    raise DataCorruption(response,
+                        f"Checksum mismatch for read_id {object_data_range.read_range.read_id}. "
+                        f"Server sent {server_checksum}, client calculated {client_checksum}."
+                    )
+
+                read_id = object_data_range.read_range.read_id
+                buffer = read_id_to_writable_buffer_dict[read_id]
+                buffer.write(data)
+                results[read_id].bytes_written += len(data)
+
+                if object_data_range.range_end:
+                    del read_id_to_writable_buffer_dict[
+                        object_data_range.read_range.read_id
+                    ]
+
+        return results
+
+    async def close(self):
+        """
+        Closes the underlying bidi-gRPC connection.
+        """
+        if not self._is_stream_open:
+            raise ValueError("Underlying bidi-gRPC stream is not open")
+        await self.read_obj_str.close()
+        self._is_stream_open = False
+
+    @property
+    def is_stream_open(self) -> bool:
+        return self._is_stream_open

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

@@ -83,32 +83,41 @@ def __init__(
         self.rpc = self.client._client._transport._wrapped_methods[
             self.client._client._transport.bidi_read_object
         ]
-        first_bidi_read_req = _storage_v2.BidiReadObjectRequest(
+        self.first_bidi_read_req = _storage_v2.BidiReadObjectRequest(
             read_object_spec=_storage_v2.BidiReadObjectSpec(
                 bucket=self._full_bucket_name, object=object_name
             ),
         )
         self.metadata = (("x-goog-request-params", f"bucket={self._full_bucket_name}"),)
-        self.socket_like_rpc = AsyncBidiRpc(
-            self.rpc, initial_request=first_bidi_read_req, metadata=self.metadata
-        )
+        self.socket_like_rpc: Optional[AsyncBidiRpc] = None
+        self._is_stream_open: bool = False
 
     async def open(self) -> None:
         """Opens the bidi-gRPC connection to read from the object.
 
         This method sends an initial request to start the stream and receives
         the first response containing metadata and a read handle.
         """
+        if self._is_stream_open:
+            raise ValueError("Stream is already open")
+        self.socket_like_rpc = AsyncBidiRpc(
+            self.rpc, initial_request=self.first_bidi_read_req, metadata=self.metadata
+        )
         await self.socket_like_rpc.open()  # this is actually 1 send
         response = await self.socket_like_rpc.recv()
         if self.generation_number is None:
             self.generation_number = response.metadata.generation
 
         self.read_handle = response.read_handle
 
+        self._is_stream_open = True
+
     async def close(self) -> None:
         """Closes the bidi-gRPC connection."""
+        if not self._is_stream_open:
+            raise ValueError("Stream is not open")
         await self.socket_like_rpc.close()
+        self._is_stream_open = False
 
     async def send(
         self, bidi_read_object_request: _storage_v2.BidiReadObjectRequest
@@ -120,6 +129,8 @@ async def send(
                 The request message to send. This is typically used to specify
                 the read offset and limit.
         """
+        if not self._is_stream_open:
+            raise ValueError("Stream is not open")
         await self.socket_like_rpc.send(bidi_read_object_request)
 
     async def recv(self) -> _storage_v2.BidiReadObjectResponse:
@@ -132,4 +143,14 @@ async def recv(self) -> _storage_v2.BidiReadObjectResponse:
             :class:`~google.cloud._storage_v2.types.BidiReadObjectResponse`:
                 The response message from the server.
         """
-        return await self.socket_like_rpc.recv()
+        if not self._is_stream_open:
+            raise ValueError("Stream is not open")
+        response = await self.socket_like_rpc.recv()
+        # Update read_handle if present in response
+        if response and response.read_handle:
+            self.read_handle = response.read_handle
+        return response
+
+    @property
+    def is_stream_open(self) -> bool:
+        return self._is_stream_open