feat(storage): Add support for blob object in AAOW (#16577)

Add support for adding custom metadata, content type and kms key name in
appendable objects

Author: nidhiii-27

packages/google-cloud-storage/google/cloud/storage/_grpc_conversions.py

@@ -0,0 +1,40 @@
+# Copyright 2026 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
+#
+#     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.
+
+from google.cloud import _storage_v2
+
+# Map Python Blob attributes to GCS V2 Object proto field names.
+_BLOB_ATTR_TO_PROTO_FIELD = {
+    "content_type": "content_type",
+    "metadata": "metadata",
+    "kms_key_name": "kms_key",
+}
+
+
+def blob_to_proto(blob):
+    """Converts a Blob instance to a GCS V2 Object proto message."""
+
+    resource_params = {
+        "name": blob.name,
+    }
+
+    if blob.bucket:
+        resource_params["bucket"] = f"projects/_/buckets/{blob.bucket.name}"
+
+    for attr_name, proto_field in _BLOB_ATTR_TO_PROTO_FIELD.items():
+        value = getattr(blob, attr_name, None)
+        if value is not None:
+            resource_params[proto_field] = value
+
+    return _storage_v2.Object(**resource_params)

packages/google-cloud-storage/google/cloud/storage/asyncio/async_appendable_object_writer.py

@@ -24,6 +24,7 @@
 from google.cloud import _storage_v2
 from google.cloud._storage_v2.types import BidiWriteObjectRedirectedError
 from google.cloud._storage_v2.types.storage import BidiWriteObjectRequest
+from google.cloud.storage import Blob
 from google.cloud.storage.asyncio.async_grpc_client import (
     AsyncGrpcClient,
 )
@@ -211,6 +212,60 @@ def __init__(
         self._routing_token: Optional[str] = None
         self.object_resource: Optional[_storage_v2.Object] = None
         self._flush_count = 0
+        self.blob: Optional[Blob] = None
+
+    @classmethod
+    def from_blob(
+        cls,
+        client: AsyncGrpcClient,
+        blob: Blob,
+        write_handle: Optional[_storage_v2.BidiWriteHandle] = None,
+        writer_options: Optional[dict] = None,
+    ) -> "AsyncAppendableObjectWriter":
+        """Creates an AsyncAppendableObjectWriter from an existing Blob object.
+
+        This factory method extracts the bucket and object names directly from
+        the provided blob instance.
+
+        .. code-block:: python
+
+            from google.cloud.storage.bucket import Bucket
+            from google.cloud.storage.blob import Blob
+
+            bucket = Bucket(client, name="my-bucket")
+            blob = Blob(name="my-object.txt", bucket=bucket)
+
+            writer = AsyncAppendableObjectWriter.from_blob(
+                client=client,
+                blob=blob
+            )
+
+        :type client: :class:`~google.cloud.storage.client.AsyncGrpcClient`
+        :param client: The async gRPC client to use for write operations.
+
+        :type blob: :class:`~google.cloud.storage.blob.Blob`
+        :param blob: The blob instance providing the target path.
+
+        :type write_handle: :class:`~google.storage.v2.BidiWriteHandle`
+        :param write_handle: (Optional) An existing BidiWriteHandle to resume a session.
+
+        :type writer_options: dict
+        :param writer_options: (Optional) Configuration settings for the underlying
+            appendable writer.
+
+        :rtype: :class:`AsyncAppendableObjectWriter`
+        :returns: An initialized writer instance.
+        """
+        instance = cls(
+            client=client,
+            bucket_name=blob.bucket.name,
+            object_name=blob.name,
+            generation=blob.generation,
+            write_handle=write_handle,
+            writer_options=writer_options,
+        )
+        instance.blob = blob
+        return instance
 
     async def state_lookup(self) -> int:
         """Returns the persisted_size
@@ -297,6 +352,7 @@ async def _do_open():
                 client=self.client.grpc_client,
                 bucket_name=self.bucket_name,
                 object_name=self.object_name,
+                blob=self.blob,
                 generation_number=self.generation,
                 write_handle=self.write_handle,
                 routing_token=self._routing_token,

packages/google-cloud-storage/google/cloud/storage/asyncio/async_write_object_stream.py

@@ -18,6 +18,8 @@
 from google.api_core.bidi_async import AsyncBidiRpc
 
 from google.cloud import _storage_v2
+from google.cloud.storage import Blob
+from google.cloud.storage import _grpc_conversions
 from google.cloud.storage.asyncio import _utils
 from google.cloud.storage.asyncio.async_abstract_object_stream import (
     _AsyncAbstractObjectStream,
@@ -67,6 +69,7 @@ def __init__(
         generation_number: Optional[int] = None,  # None means new object
         write_handle: Optional[_storage_v2.BidiWriteHandle] = None,
         routing_token: Optional[str] = None,
+        blob: Optional[Blob] = None,
     ) -> None:
         if client is None:
             raise ValueError("client must be provided")
@@ -83,7 +86,7 @@ def __init__(
         self.client: AsyncGrpcClient.grpc_client = client
         self.write_handle: Optional[_storage_v2.BidiWriteHandle] = write_handle
         self.routing_token: Optional[str] = routing_token
-
+        self.blob: Optional[Blob] = blob
         self._full_bucket_name = f"projects/_/buckets/{self.bucket_name}"
 
         self.rpc = self.client._client._transport._wrapped_methods[
@@ -118,11 +121,15 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> 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:
+            if self.blob:
+                resource = _grpc_conversions.blob_to_proto(self.blob)
+            else:
+                resource = _storage_v2.Object(
+                    name=self.object_name, bucket=self._full_bucket_name
+                )
             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
-                    ),
+                    resource=resource,
                     appendable=True,
                     if_generation_match=self.generation_number,
                 ),

packages/google-cloud-storage/tests/system/test_zonal.py

@@ -17,6 +17,7 @@
 )
 
 # current library imports
+from google.cloud import kms
 from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient
 from google.cloud.storage.asyncio.async_multi_range_downloader import (
     AsyncMultiRangeDownloader,
@@ -29,7 +30,7 @@
 
 
 # TODO: replace this with a fixture once zonal bucket creation / deletion
-# is supported in grpc client or json client client.
+# is supported in grpc client or json client.
 _ZONAL_BUCKET = os.getenv("ZONAL_BUCKET")
 _CROSS_REGION_BUCKET = os.getenv("CROSS_REGION_BUCKET")
 _BYTES_TO_UPLOAD = b"dummy_bytes_to_write_read_and_delete_appendable_object"
@@ -40,6 +41,58 @@ async def create_async_grpc_client(attempt_direct_path=True):
     return AsyncGrpcClient(attempt_direct_path=attempt_direct_path)
 
 
+@pytest.fixture(scope="session")
+def zonal_kms_key(storage_client, kms_client):
+    """Provisions a KMS key in the same location as of the zonal bucket."""
+    # Get the zonal bucket and extract its location
+    bucket = storage_client.get_bucket(_ZONAL_BUCKET)
+    location = bucket.location.lower()
+
+    project = storage_client.project
+    keyring_name = "gcs-test-zonal-ring"
+    key_name = "gcs-test-zonal-key"
+
+    keyring_path = kms_client.key_ring_path(project, location, keyring_name)
+
+    # Create the KeyRing if it doesn't exist
+    try:
+        kms_client.get_key_ring(name=keyring_path)
+    except NotFound:
+        parent = f"projects/{project}/locations/{location}"
+        kms_client.create_key_ring(
+            request={"parent": parent, "key_ring_id": keyring_name, "key_ring": {}}
+        )
+
+        # Grant GCS service account permissions to use the key
+        service_account_email = storage_client.get_service_account_email()
+        policy = {
+            "bindings": [
+                {
+                    "role": "roles/cloudkms.cryptoKeyEncrypterDecrypter",
+                    "members": [f"serviceAccount:{service_account_email}"],
+                }
+            ]
+        }
+        kms_client.set_iam_policy(request={"resource": keyring_path, "policy": policy})
+
+    # Create the CryptoKey if it doesn't exist
+    key_path = kms_client.crypto_key_path(project, location, keyring_name, key_name)
+    try:
+        kms_client.get_crypto_key(name=key_path)
+    except NotFound:
+        purpose = kms.CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT
+        key = {"purpose": purpose}
+        kms_client.create_crypto_key(
+            request={
+                "parent": keyring_path,
+                "crypto_key_id": key_name,
+                "crypto_key": key,
+            }
+        )
+
+    return key_path
+
+
 @pytest.fixture(scope="session")
 def event_loop():
     """Redefine pytest-asyncio's event_loop fixture to be session-scoped."""
@@ -286,6 +339,83 @@ async def _run():
     event_loop.run_until_complete(_run())
 
 
+def test_write_from_blob(
+    storage_client,
+    blobs_to_delete,
+    event_loop,
+    grpc_client,
+):
+    object_name = f"test_from_blob-{str(uuid.uuid4())[:4]}"
+    content_type = "text/plain"
+    metadata = {"environment": "system-test"}
+    test_data = b"system-test-data"
+
+    async def _run():
+        # 1. Create a Blob instance
+        blob = storage_client.bucket(_ZONAL_BUCKET).blob(object_name)
+        blob.content_type = content_type
+        blob.metadata = metadata
+
+        # 2. Use from_blob to create the writer
+        writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob)
+        await writer.open()
+        await writer.append(test_data)
+        await writer.close(finalize_on_close=True)
+
+        # 3. Verify the object metadata
+        obj = await grpc_client.get_object(
+            bucket_name=_ZONAL_BUCKET,
+            object_name=object_name,
+        )
+
+        assert obj.content_type == content_type
+        assert obj.metadata["environment"] == "system-test"
+
+        blobs_to_delete.append(blob)
+
+    event_loop.run_until_complete(_run())
+
+
+def test_write_from_blob_with_kms_key(
+    storage_client,
+    blobs_to_delete,
+    event_loop,
+    grpc_client,
+    zonal_kms_key,
+):
+    """Verifies AsyncAppendableObjectWriter.from_blob correctly applies KMS encryption."""
+
+    object_name = f"test_from_blob_kms-{str(uuid.uuid4())[:4]}"
+    test_data = b"kms-protected-data"
+
+    async def _run():
+        # Create a local Blob instance with the KMS key
+        blob = storage_client.bucket(_ZONAL_BUCKET).blob(
+            object_name, kms_key_name=zonal_kms_key
+        )
+
+        writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob)
+
+        await writer.open()
+        await writer.append(test_data)
+
+        await writer.close(finalize_on_close=True)
+
+        # Verify the encryption metadata
+        obj = await grpc_client.get_object(
+            bucket_name=_ZONAL_BUCKET,
+            object_name=object_name,
+        )
+
+        # Assert that the object was encrypted with the correct key
+        # GCS appends a version suffix, so we use startswith()
+        assert obj.kms_key.startswith(zonal_kms_key)
+
+        blobs_to_delete.append(blob)
+
+    event_loop.run_until_complete(_run())
+
+
 def test_read_unfinalized_appendable_object(
     storage_client, blobs_to_delete, event_loop, grpc_client_direct
 ):

packages/google-cloud-storage/tests/unit/asyncio/test_async_appendable_object_writer.py

@@ -19,6 +19,7 @@
 import pytest
 from google.api_core import exceptions
 from google.rpc import status_pb2
+from google.cloud.storage import Blob
 
 from google.cloud._storage_v2.types import storage as storage_type
 from google.cloud._storage_v2.types.storage import BidiWriteObjectRedirectedError
@@ -166,6 +167,24 @@ def test_init_raises_if_crc32c_missing(self, mock_appendable_writer):
             with pytest.raises(exceptions.FailedPrecondition):
                 self._make_one(mock_appendable_writer["mock_client"])
 
+    def test_from_blob(self, mock_appendable_writer):
+        mock_blob = mock.Mock(spec=Blob)
+        mock_blob.name = OBJECT
+        mock_blob.bucket.name = BUCKET
+        mock_blob.generation = GENERATION
+
+        writer = AsyncAppendableObjectWriter.from_blob(
+            mock_appendable_writer["mock_client"],
+            mock_blob,
+            writer_options={"FLUSH_INTERVAL_BYTES": EIGHT_MIB},
+        )
+
+        assert writer.bucket_name == BUCKET
+        assert writer.object_name == OBJECT
+        assert writer.generation == GENERATION
+        assert writer.flush_interval == EIGHT_MIB
+        assert writer.blob == mock_blob
+
     # -------------------------------------------------------------------------
     # Stream Lifecycle Tests
     # -------------------------------------------------------------------------