feat: add fixed-key metadata support in AAOW (#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.
 
 fromgoogle.cloudimport_storage_v2
+fromgoogle.protobufimporttimestamp_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):
 ifvalueisnotNone:
 resource_params[proto_field] =value
 
+custom_time=getattr(blob, "custom_time", None)
+ifcustom_timeisnotNone:
+custom_time_proto=timestamp_pb2.Timestamp()
+custom_time_proto.FromDatetime(custom_time)
+resource_params["custom_time"] =custom_time_proto
+
+acl=getattr(blob, "acl", None)
+ifaclisnotNoneandgetattr(acl, "loaded", False):
+acl_entries= []
+forentryinacl:
+acl_entries.append(
+_storage_v2.ObjectAccessControl(
+role=entry["role"],
+entity=entry["entity"],
+ )
+ )
+ifacl_entries:
+resource_params["acl"] =acl_entries
+
+retention=getattr(blob, "retention", None)
+ifretention:
+mode_str=retention.get("mode")
+mode=_storage_v2.Object.Retention.Mode.MODE_UNSPECIFIED
+ifmode_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")
+ifretain_until_timeisnotNone:
+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
 importasyncio
+importdatetime
 importgc
 importos
 importrandom
@@ -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"
 
 asyncdef_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():
 
 assertobj.content_type==content_type
 assertobj.metadata["environment"] =="system-test"
+assertobj.cache_control==cache_control
+assertobj.content_disposition==content_disposition
+assertobj.content_encoding==content_encoding
+assertobj.content_language==content_language
+assertint(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.
 
+importdatetime
 importunittest.mockasmock
 fromunittest.mockimportAsyncMock, 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)
 awaitstream.open()
@@ -180,6 +207,24 @@ async def test_open_new_object_with_blob_sync_attrs(
 
 assertresource.content_type=="text/plain"
 assertresource.metadata== {"test-key": "test-value"}
+assertresource.kms_key=="kms-key-name"
+assertresource.cache_control=="cache-control"
+assertresource.content_disposition=="content-disposition"
+assertresource.content_encoding=="content-encoding"
+assertresource.content_language=="content-language"
+assertresource.temporary_holdisTrue
+assertresource.event_based_holdisTrue
+
+assertint(resource.custom_time.timestamp()) ==int(custom_time.timestamp())
+
+assertlen(resource.acl) ==1
+assertresource.acl[0].role=="READER"
+assertresource.acl[0].entity=="allUsers"
+
+assertresource.retention.mode==_storage_v2.Object.Retention.Mode.LOCKED
+assertint(resource.retention.retain_until_time.timestamp()) ==int(
+retain_until_time.timestamp()
+ )
 
 @pytest.mark.asyncio
 asyncdeftest_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.
+
+importdatetime
+fromunittestimportmock
+
+fromgoogle.cloud.storageimport_grpc_conversions
+fromgoogle.cloudimport_storage_v2
+
+
+deftest_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)
+
+assertproto.name=="blob-name"
+assertproto.bucket=="projects/_/buckets/bucket-name"
+assertproto.content_type=="text/plain"
+assertproto.metadata== {"key": "value"}
+assertproto.kms_key=="kms-key"
+assertproto.cache_control=="no-cache"
+assertproto.content_disposition=="attachment"
+assertproto.content_encoding=="gzip"
+assertproto.content_language=="en"
+assertproto.temporary_holdisTrue
+assertproto.event_based_holdisFalse
+
+
+deftest_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
+forattrin_grpc_conversions._BLOB_ATTR_TO_PROTO_FIELD:
+setattr(blob, attr, None)
+
+proto=_grpc_conversions.blob_to_proto(blob)
+
+assertint(proto.custom_time.timestamp()) ==int(blob.custom_time.timestamp())
+
+
+deftest_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
+forattrin_grpc_conversions._BLOB_ATTR_TO_PROTO_FIELD:
+setattr(blob, attr, None)
+
+proto=_grpc_conversions.blob_to_proto(blob)
+
+assertlen(proto.acl) ==2
+assertproto.acl[0].role=="READER"
+assertproto.acl[0].entity=="allUsers"
+assertproto.acl[1].role=="OWNER"
+assertproto.acl[1].entity=="user-123"
+
+
+deftest_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
+forattrin_grpc_conversions._BLOB_ATTR_TO_PROTO_FIELD:
+setattr(blob, attr, None)
+
+proto=_grpc_conversions.blob_to_proto(blob)
+
+assertproto.retention.mode==_storage_v2.Object.Retention.Mode.LOCKED
+assertint(proto.retention.retain_until_time.timestamp()) ==int(
+retain_until_time.timestamp()
+ )