feat: add fixed-key metadata support in AAOW (googleapis#16817)

Add fixed-key metadata support in AAOW by updating `blob_to_proto`
conversion logic.
- Updated `_grpc_conversions.py` with simple and complex field mappings.
- Added unit tests in `tests/unit/test__grpc_conversions.py`.
- Updated system tests in `tests/system/test_zonal.py`.
- Fixed regression in
`tests/unit/asyncio/test_async_write_object_stream.py`.

---
*PR created automatically by Jules for task
[11384837182247010380](https://jules.google.com/task/11384837182247010380)
started by @nidhiii-27*

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: nidhiii-27 <224584462+nidhiii-27@users.noreply.github.com>

Author: nidhiii-27

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

@@ -13,12 +13,19 @@
 # limitations under the License.
 
 from google.cloud import _storage_v2
+from google.protobuf import timestamp_pb2
 
 # 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",
+    "cache_control": "cache_control",
+    "content_disposition": "content_disposition",
+    "content_encoding": "content_encoding",
+    "content_language": "content_language",
+    "temporary_hold": "temporary_hold",
+    "event_based_hold": "event_based_hold",
 }
 
 
@@ -37,4 +44,46 @@ def blob_to_proto(blob):
         if value is not None:
             resource_params[proto_field] = value
 
+    custom_time = getattr(blob, "custom_time", None)
+    if custom_time is not None:
+        custom_time_proto = timestamp_pb2.Timestamp()
+        custom_time_proto.FromDatetime(custom_time)
+        resource_params["custom_time"] = custom_time_proto
+
+    acl = getattr(blob, "acl", None)
+    if acl is not None and getattr(acl, "loaded", False):
+        acl_entries = []
+        for entry in acl:
+            acl_entries.append(
+                _storage_v2.ObjectAccessControl(
+                    role=entry["role"],
+                    entity=entry["entity"],
+                )
+            )
+        if acl_entries:
+            resource_params["acl"] = acl_entries
+
+    retention = getattr(blob, "retention", None)
+    if retention:
+        mode_str = retention.get("mode")
+        mode = _storage_v2.Object.Retention.Mode.MODE_UNSPECIFIED
+        if mode_str:
+            # GCS retention modes are 'Locked' or 'Unlocked'
+            mode = getattr(
+                _storage_v2.Object.Retention.Mode,
+                mode_str.upper(),
+                _storage_v2.Object.Retention.Mode.MODE_UNSPECIFIED,
+            )
+
+        retain_until_time_proto = None
+        retain_until_time = retention.get("retain_until_time")
+        if retain_until_time is not None:
+            retain_until_time_proto = timestamp_pb2.Timestamp()
+            retain_until_time_proto.FromDatetime(retain_until_time)
+
+        resource_params["retention"] = _storage_v2.Object.Retention(
+            mode=mode,
+            retain_until_time=retain_until_time_proto,
+        )
+
     return _storage_v2.Object(**resource_params)

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

@@ -1,5 +1,6 @@
 # py standard imports
 import asyncio
+import datetime
 import gc
 import os
 import random
@@ -347,13 +348,23 @@ def test_write_from_blob(
     object_name = f"test_from_blob-{str(uuid.uuid4())[:4]}"
     content_type = "text/plain"
     metadata = {"environment": "system-test"}
+    cache_control = "public, max-age=3600"
+    content_disposition = "attachment; filename=test.txt"
+    content_encoding = "identity"
+    content_language = "en"
+    custom_time = datetime.datetime(2025, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc)
     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
+        blob.cache_control = cache_control
+        blob.content_disposition = content_disposition
+        blob.content_encoding = content_encoding
+        blob.content_language = content_language
+        blob.custom_time = custom_time
 
         # 2. Use from_blob to create the writer
         writer = AsyncAppendableObjectWriter.from_blob(grpc_client, blob)
@@ -369,6 +380,11 @@ async def _run():
 
         assert obj.content_type == content_type
         assert obj.metadata["environment"] == "system-test"
+        assert obj.cache_control == cache_control
+        assert obj.content_disposition == content_disposition
+        assert obj.content_encoding == content_encoding
+        assert obj.content_language == content_language
+        assert int(obj.custom_time.timestamp()) == int(custom_time.timestamp())
 
         blobs_to_delete.append(blob)
 

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

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import datetime
 import unittest.mock as mock
 from unittest.mock import AsyncMock, MagicMock
 
@@ -169,7 +170,33 @@ async def test_open_new_object_with_blob_sync_attrs(
         mock_blob.bucket = mock_bucket
         mock_blob.content_type = "text/plain"
         mock_blob.metadata = {"test-key": "test-value"}
-        mock_blob.kms_key_name = None
+        mock_blob.kms_key_name = "kms-key-name"
+        mock_blob.cache_control = "cache-control"
+        mock_blob.content_disposition = "content-disposition"
+        mock_blob.content_encoding = "content-encoding"
+        mock_blob.content_language = "content-language"
+        mock_blob.temporary_hold = True
+        mock_blob.event_based_hold = True
+
+        custom_time = datetime.datetime(
+            2025, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc
+        )
+        mock_blob.custom_time = custom_time
+
+        acl_mock = MagicMock()
+        acl_mock.loaded = True
+        acl_mock.__iter__.return_value = iter(
+            [{"role": "READER", "entity": "allUsers"}]
+        )
+        mock_blob.acl = acl_mock
+
+        retain_until_time = datetime.datetime(
+            2026, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc
+        )
+        mock_blob.retention = {
+            "mode": "Locked",
+            "retain_until_time": retain_until_time,
+        }
 
         stream = _AsyncWriteObjectStream(mock_client, BUCKET, OBJECT, blob=mock_blob)
         await stream.open()
@@ -180,6 +207,24 @@ async def test_open_new_object_with_blob_sync_attrs(
 
         assert resource.content_type == "text/plain"
         assert resource.metadata == {"test-key": "test-value"}
+        assert resource.kms_key == "kms-key-name"
+        assert resource.cache_control == "cache-control"
+        assert resource.content_disposition == "content-disposition"
+        assert resource.content_encoding == "content-encoding"
+        assert resource.content_language == "content-language"
+        assert resource.temporary_hold is True
+        assert resource.event_based_hold is True
+
+        assert int(resource.custom_time.timestamp()) == int(custom_time.timestamp())
+
+        assert len(resource.acl) == 1
+        assert resource.acl[0].role == "READER"
+        assert resource.acl[0].entity == "allUsers"
+
+        assert resource.retention.mode == _storage_v2.Object.Retention.Mode.LOCKED
+        assert int(resource.retention.retain_until_time.timestamp()) == int(
+            retain_until_time.timestamp()
+        )
 
     @pytest.mark.asyncio
     async def test_open_already_open_raises(self, mock_client):

packages/google-cloud-storage/tests/unit/test__grpc_conversions.py

@@ -0,0 +1,136 @@
+# 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.
+
+import datetime
+from unittest import mock
+
+from google.cloud.storage import _grpc_conversions
+from google.cloud import _storage_v2
+
+
+def test_blob_to_proto_simple_fields():
+    blob = mock.Mock(
+        spec=[
+            "name",
+            "bucket",
+            "content_type",
+            "metadata",
+            "kms_key_name",
+            "cache_control",
+            "content_disposition",
+            "content_encoding",
+            "content_language",
+            "temporary_hold",
+            "event_based_hold",
+            "custom_time",
+            "acl",
+            "retention",
+        ]
+    )
+    blob.name = "blob-name"
+    blob.bucket.name = "bucket-name"
+    blob.content_type = "text/plain"
+    blob.metadata = {"key": "value"}
+    blob.kms_key_name = "kms-key"
+    blob.cache_control = "no-cache"
+    blob.content_disposition = "attachment"
+    blob.content_encoding = "gzip"
+    blob.content_language = "en"
+    blob.temporary_hold = True
+    blob.event_based_hold = False
+    blob.custom_time = None
+    blob.acl = None
+    blob.retention = None
+
+    proto = _grpc_conversions.blob_to_proto(blob)
+
+    assert proto.name == "blob-name"
+    assert proto.bucket == "projects/_/buckets/bucket-name"
+    assert proto.content_type == "text/plain"
+    assert proto.metadata == {"key": "value"}
+    assert proto.kms_key == "kms-key"
+    assert proto.cache_control == "no-cache"
+    assert proto.content_disposition == "attachment"
+    assert proto.content_encoding == "gzip"
+    assert proto.content_language == "en"
+    assert proto.temporary_hold is True
+    assert proto.event_based_hold is False
+
+
+def test_blob_to_proto_custom_time():
+    blob = mock.Mock(spec=["name", "bucket", "custom_time", "acl", "retention"])
+    blob.name = "blob-name"
+    blob.bucket.name = "bucket-name"
+    blob.custom_time = datetime.datetime(
+        2025, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc
+    )
+    blob.acl = None
+    blob.retention = None
+    # ensure other fields don't cause issues if missing
+    for attr in _grpc_conversions._BLOB_ATTR_TO_PROTO_FIELD:
+        setattr(blob, attr, None)
+
+    proto = _grpc_conversions.blob_to_proto(blob)
+
+    assert int(proto.custom_time.timestamp()) == int(blob.custom_time.timestamp())
+
+
+def test_blob_to_proto_acl():
+    blob = mock.Mock(spec=["name", "bucket", "acl", "custom_time", "retention"])
+    blob.name = "blob-name"
+    blob.bucket.name = "bucket-name"
+
+    acl_mock = mock.MagicMock()
+    acl_mock.loaded = True
+    acl_mock.__iter__.return_value = iter(
+        [
+            {"role": "READER", "entity": "allUsers"},
+            {"role": "OWNER", "entity": "user-123"},
+        ]
+    )
+    blob.acl = acl_mock
+
+    blob.custom_time = None
+    blob.retention = None
+    for attr in _grpc_conversions._BLOB_ATTR_TO_PROTO_FIELD:
+        setattr(blob, attr, None)
+
+    proto = _grpc_conversions.blob_to_proto(blob)
+
+    assert len(proto.acl) == 2
+    assert proto.acl[0].role == "READER"
+    assert proto.acl[0].entity == "allUsers"
+    assert proto.acl[1].role == "OWNER"
+    assert proto.acl[1].entity == "user-123"
+
+
+def test_blob_to_proto_retention():
+    blob = mock.Mock(spec=["name", "bucket", "retention", "custom_time", "acl"])
+    blob.name = "blob-name"
+    blob.bucket.name = "bucket-name"
+
+    retain_until_time = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+    blob.retention = {"mode": "Locked", "retain_until_time": retain_until_time}
+
+    blob.custom_time = None
+    blob.acl = None
+    for attr in _grpc_conversions._BLOB_ATTR_TO_PROTO_FIELD:
+        setattr(blob, attr, None)
+
+    proto = _grpc_conversions.blob_to_proto(blob)
+
+    assert proto.retention.mode == _storage_v2.Object.Retention.Mode.LOCKED
+    assert int(proto.retention.retain_until_time.timestamp()) == int(
+        retain_until_time.timestamp()
+    )