Merge branch 'main' into bench

Author: chandra-siri

cloudbuild/run_zonal_tests.sh

@@ -26,3 +26,4 @@ export RUN_ZONAL_SYSTEM_TESTS=True
 CURRENT_ULIMIT=$(ulimit -n)
 echo '--- Running Zonal tests on VM with ulimit set to ---' $CURRENT_ULIMIT
 pytest -vv -s --log-format='%(asctime)s %(levelname)s %(message)s' --log-date-format='%H:%M:%S' tests/system/test_zonal.py
+pytest -vv -s --log-format='%(asctime)s %(levelname)s %(message)s' --log-date-format='%H:%M:%S' samples/snippets/zonal_buckets/zonal_snippets_test.py

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

@@ -97,15 +97,29 @@ def __init__(
         :param object_name: The name of the GCS Appendable Object to be written.
 
         :type generation: int
-        :param generation: (Optional) If present, selects a specific revision of
-                            that object.
-                            If None, a new object is created.
-                            If None and Object already exists then it'll will be
-                            overwritten.
+        :param generation: (Optional) If present, creates writer for that
+            specific revision of that object. Use this to append data to an
+            existing Appendable Object.
+
+            Setting to ``0`` makes the `writer.open()` succeed only if
+            object doesn't exist in the bucket (useful for not accidentally
+            overwriting existing objects).
+
+            Warning: If `None`, a new object is created. If an object with the
+            same name already exists, it will be overwritten the moment 
+            `writer.open()` is called.
 
         :type write_handle: bytes
-        :param write_handle: (Optional) An existing handle for writing the object.
-                            If provided, opening the bidi-gRPC connection will be faster.
+        :param write_handle: (Optional) An handle for writing the object.
+            If provided, opening the bidi-gRPC connection will be faster.
+
+        :type writer_options: dict
+        :param writer_options: (Optional) A dictionary of writer options.
+            Supported options:
+            - "FLUSH_INTERVAL_BYTES": int
+                The number of bytes to append before "persisting" data in GCS
+                servers. Default is `_DEFAULT_FLUSH_INTERVAL_BYTES`.
+                Must be a multiple of `_MAX_CHUNK_SIZE_BYTES`.
         """
         raise_if_no_fast_crc32c()
         self.client = client
@@ -133,7 +147,6 @@ def __init__(
         self.flush_interval = writer_options.get(
             "FLUSH_INTERVAL_BYTES", _DEFAULT_FLUSH_INTERVAL_BYTES
         )
-        # TODO: add test case for this.
         if self.flush_interval < _MAX_CHUNK_SIZE_BYTES:
             raise exceptions.OutOfRange(
                 f"flush_interval must be >= {_MAX_CHUNK_SIZE_BYTES} , but provided {self.flush_interval}"
@@ -346,6 +359,11 @@ async def finalize(self) -> _storage_v2.Object:
         self.offset = None
         return self.object_resource
 
+    @property
+    def is_stream_open(self) -> bool:
+        return self._is_stream_open
+
+
     # helper methods.
     async def append_from_string(self, data: str):
         """

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

@@ -15,6 +15,9 @@
 """An async client for interacting with Google Cloud Storage using the gRPC API."""
 
 from google.cloud import _storage_v2 as storage_v2
+from google.cloud._storage_v2.services.storage.transports.base import (
+    DEFAULT_CLIENT_INFO,
+)
 
 
 class AsyncGrpcClient:
@@ -30,7 +33,7 @@ class AsyncGrpcClient:
         The client info used to send a user-agent string along with API
         requests. If ``None``, then default info will be used.
 
-    :type client_options: :class:`~google.api_core.client_options.ClientOptions` or :class:`dict`
+    :type client_options: :class:`~google.api_core.client_options.ClientOptions`
     :param client_options: (Optional) Client options used to set user options
         on the client.
 
@@ -39,7 +42,6 @@ class AsyncGrpcClient:
         (Optional) Whether to attempt to use DirectPath for gRPC connections.
         Defaults to ``True``.
     """
-
     def __init__(
         self,
         credentials=None,
@@ -65,8 +67,15 @@ def _create_async_grpc_client(
         transport_cls = storage_v2.StorageAsyncClient.get_transport_class(
             "grpc_asyncio"
         )
+
+        if client_info is None:
+            client_info = DEFAULT_CLIENT_INFO
+        primary_user_agent = client_info.to_user_agent()
+
         channel = transport_cls.create_channel(
-            attempt_direct_path=attempt_direct_path, credentials=credentials
+            attempt_direct_path=attempt_direct_path,
+            credentials=credentials,
+            options=(("grpc.primary_user_agent", primary_user_agent),),
         )
         transport = transport_cls(channel=channel)
 

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

@@ -303,11 +303,19 @@ async def download_ranges(
 
         :type read_ranges: List[Tuple[int, int, "BytesIO"]]
         :param read_ranges: A list of tuples, where each tuple represents a
-            combintaion of byte_range and writeable buffer in format -
+            combination of byte_range and writeable buffer in format -
             (`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.
 
+            Special cases:
+            if the value of `bytes_to_read` is 0, it'll be interpreted as
+            download all contents until the end of the file from `start_byte`.
+            Examples:
+                * (0, 0, buffer) : downloads 0 to end , i.e. entire object.
+                * (100, 0, buffer) : downloads from 100 to end.
+
+
         :type lock: asyncio.Lock
         :param lock: (Optional) An asyncio lock to synchronize sends and recvs
             on the underlying bidi-GRPC stream. This is required when multiple

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

@@ -151,9 +151,16 @@ async def close(self) -> None:
         """Closes the bidi-gRPC connection."""
         if not self._is_stream_open:
             raise ValueError("Stream is not open")
+        await self.requests_done()
         await self.socket_like_rpc.close()
         self._is_stream_open = False
 
+    async def requests_done(self):
+        """Signals that all requests have been sent."""
+
+        await self.socket_like_rpc.send(None)
+        await self.socket_like_rpc.recv()
+
     async def send(
         self, bidi_read_object_request: _storage_v2.BidiReadObjectRequest
     ) -> None:

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

@@ -47,8 +47,17 @@ class _AsyncWriteObjectStream(_AsyncAbstractObjectStream):
     :param object_name: The name of the GCS ``Appendable Object`` to be write.
 
     :type generation_number: int
-    :param generation_number: (Optional) If present, selects a specific revision of
-                              this object. If None, a new object is created.
+    :param generation_number: (Optional) If present, creates writer for that
+        specific revision of that object. Use this to append data to an
+        existing Appendable Object.
+
+        Setting to ``0`` makes the `writer.open()` succeed only if
+        object doesn't exist in the bucket (useful for not accidentally
+        overwriting existing objects).
+
+        Warning: If `None`, a new object is created. If an object with the
+        same name already exists, it will be overwritten the moment
+        `writer.open()` is called.
 
     :type write_handle: bytes
     :param write_handle: (Optional) An existing handle for writing the object.
@@ -92,22 +101,33 @@ def __init__(
         self.object_resource: Optional[_storage_v2.Object] = None
 
     async def open(self) -> None:
-        """Opening an object for write , should do it's state lookup
-        to know what's the persisted size is.
+        """
+        Opens the bidi-gRPC connection to write to the object.
+
+        This method sends an initial request to start the stream and receives
+        the first response containing metadata and a write handle.
+
+        :rtype: None
+        :raises ValueError: If the stream is already open.
+        :raises google.api_core.exceptions.FailedPrecondition: 
+            if `generation_number` is 0 and object already exists.
         """
         if self._is_stream_open:
             raise ValueError("Stream is already open")
 
         # Create a new object or overwrite existing one if generation_number
         # is None. This makes it consistent with GCS JSON API behavior.
         # Created object type would be Appendable Object.
-        if self.generation_number is None:
+        # if `generation_number` == 0 new object will be created only if there
+        # isn't any existing object.
+        if self.generation_number is None or self.generation_number == 0:
             self.first_bidi_write_req = _storage_v2.BidiWriteObjectRequest(
                 write_object_spec=_storage_v2.WriteObjectSpec(
                     resource=_storage_v2.Object(
                         name=self.object_name, bucket=self._full_bucket_name
                     ),
                     appendable=True,
+                    if_generation_match=self.generation_number,
                 ),
             )
         else:
@@ -118,7 +138,6 @@ async def open(self) -> None:
                     generation=self.generation_number,
                 ),
             )
-
         self.socket_like_rpc = AsyncBidiRpc(
             self.rpc, initial_request=self.first_bidi_write_req, metadata=self.metadata
         )
@@ -152,9 +171,16 @@ async def close(self) -> None:
         """Closes the bidi-gRPC connection."""
         if not self._is_stream_open:
             raise ValueError("Stream is not open")
+        await self.requests_done()
         await self.socket_like_rpc.close()
         self._is_stream_open = False
 
+    async def requests_done(self):
+        """Signals that all requests have been sent."""
+
+        await self.socket_like_rpc.send(None)
+        await self.socket_like_rpc.recv()
+
     async def send(
         self, bidi_write_object_request: _storage_v2.BidiWriteObjectRequest
     ) -> None:
@@ -186,3 +212,4 @@ async def recv(self) -> _storage_v2.BidiWriteObjectResponse:
     @property
     def is_stream_open(self) -> bool:
         return self._is_stream_open
+

samples/snippets/zonal_buckets/storage_create_and_write_appendable_object.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python
+
+# Copyright 2026 Google Inc. All Rights Reserved.
+#
+# 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
+#
+#     http://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 argparse
+import asyncio
+
+from google.cloud.storage._experimental.asyncio.async_appendable_object_writer import (
+    AsyncAppendableObjectWriter,
+)
+from google.cloud.storage._experimental.asyncio.async_grpc_client import AsyncGrpcClient
+
+
+# [START storage_create_and_write_appendable_object]
+
+
+async def storage_create_and_write_appendable_object(
+    bucket_name, object_name, grpc_client=None
+):
+    """Uploads an appendable object to zonal bucket.
+
+    grpc_client: an existing grpc_client to use, this is only for testing.
+    """
+
+    if grpc_client is None:
+        grpc_client = AsyncGrpcClient().grpc_client
+    writer = AsyncAppendableObjectWriter(
+        client=grpc_client,
+        bucket_name=bucket_name,
+        object_name=object_name,
+        generation=0,  # throws `FailedPrecondition` if object already exists.
+    )
+    # This creates a new appendable object of size 0 and opens it for appending.
+    await writer.open()
+
+    # appends data to the object
+    # you can perform `.append` multiple times as needed. Data will be appended
+    # to the end of the object.
+    await writer.append(b"Some data")
+
+    # Once all appends are done, close the gRPC bidirectional stream.
+    await writer.close()
+
+    print(
+        f"Appended object {object_name} created of size {writer.persisted_size} bytes."
+    )
+
+
+# [END storage_create_and_write_appendable_object]
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    parser.add_argument("--bucket_name", help="Your Cloud Storage bucket name.")
+    parser.add_argument("--object_name", help="Your Cloud Storage object name.")
+
+    args = parser.parse_args()
+
+    asyncio.run(
+        storage_create_and_write_appendable_object(
+            bucket_name=args.bucket_name,
+            object_name=args.object_name,
+        )
+    )

samples/snippets/zonal_buckets/storage_finalize_appendable_object_upload.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python
+
+# Copyright 2026 Google Inc. All Rights Reserved.
+#
+# 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
+#
+#     http://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 argparse
+import asyncio
+
+from google.cloud.storage._experimental.asyncio.async_appendable_object_writer import (
+    AsyncAppendableObjectWriter,
+)
+from google.cloud.storage._experimental.asyncio.async_grpc_client import AsyncGrpcClient
+
+
+# [START storage_finalize_appendable_object_upload]
+async def storage_finalize_appendable_object_upload(
+    bucket_name, object_name, grpc_client=None
+):
+    """Creates, writes to, and finalizes an appendable object.
+
+    grpc_client: an existing grpc_client to use, this is only for testing.
+    """
+
+    if grpc_client is None:
+        grpc_client = AsyncGrpcClient().grpc_client
+    writer = AsyncAppendableObjectWriter(
+        client=grpc_client,
+        bucket_name=bucket_name,
+        object_name=object_name,
+        generation=0,  # throws `FailedPrecondition` if object already exists.
+    )
+    # This creates a new appendable object of size 0 and opens it for appending.
+    await writer.open()
+
+    # Appends data to the object.
+    await writer.append(b"Some data")
+
+    # finalize the appendable object,
+    # NOTE:
+    # 1. once finalized no more appends can be done to the object.
+    # 2. If you don't want to finalize, you can simply call `writer.close`
+    # 3. calling `.finalize()` also closes the grpc-bidi stream, calling
+    #   `.close` after `.finalize` may lead to undefined behavior.
+    object_resource = await writer.finalize()
+
+    print(f"Appendable object {object_name} created and finalized.")
+    print("Object Metadata:")
+    print(object_resource)
+
+
+# [END storage_finalize_appendable_object_upload]
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    parser.add_argument("--bucket_name", help="Your Cloud Storage bucket name.")
+    parser.add_argument("--object_name", help="Your Cloud Storage object name.")
+
+    args = parser.parse_args()
+
+    asyncio.run(
+        storage_finalize_appendable_object_upload(
+            bucket_name=args.bucket_name,
+            object_name=args.object_name,
+        )
+    )