removes metadata from reader and writer | updates docstring

Author: ankitaluthra1

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

@@ -213,7 +213,6 @@ def __init__(
         self.object_resource: Optional[_storage_v2.Object] = None
         self._flush_count = 0
         self.blob: Optional[Blob] = None
-        self.metadata: Optional[List[Tuple[str, str]]] = None
 
     @classmethod
     def from_blob(
@@ -313,8 +312,6 @@ async def open(
         if self._is_stream_open:
             raise ValueError("Underlying bidi-gRPC stream is already open")
 
-        self.metadata = metadata
-
         if retry_policy is None:
             retry_policy = AsyncRetry(
                 predicate=_is_write_retryable, on_error=self._on_open_error
@@ -337,7 +334,7 @@ def combined_on_error(exc):
             )
 
         async def _do_open():
-            current_metadata = list(self.metadata) if self.metadata else []
+            current_metadata = list(metadata) if metadata else []
 
             # Cleanup stream from previous failed attempt, if any.
             if self.write_obj_stream:
@@ -411,11 +408,6 @@ async def append(
 
         :raises ValueError: If the stream is not open.
         """
-        if metadata is not None:
-            self.metadata = metadata
-        else:
-            metadata = self.metadata
-
         if not self._is_stream_open:
             raise ValueError("Stream is not open. Call open() before append().")
         if not data:

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

@@ -26,6 +26,22 @@
 _DEFAULT_HOST = "storage.googleapis.com"
 
 
+def _validate_metadata(metadata):
+    """Validates that metadata is a sequence of (key, value) pairs."""
+    if metadata is None:
+        return
+    if not isinstance(metadata, (list, tuple)):
+        raise TypeError("metadata must be a list or tuple of (key, value) pairs.")
+    for item in metadata:
+        if not isinstance(item, (list, tuple)) or len(item) != 2:
+            raise ValueError(
+                "Each element in metadata must be a list or tuple of exactly 2 strings (key, value)."
+            )
+        key, value = item
+        if not isinstance(key, str) or not isinstance(value, str):
+            raise TypeError("Both key and value in metadata pairs must be strings.")
+
+
 class AsyncGrpcClient:
     """An asynchronous client for interacting with Google Cloud Storage using the gRPC API.
 
@@ -184,8 +200,18 @@ async def delete_object(
         :type if_metageneration_not_match: int
         :param if_metageneration_not_match: (Optional)
 
+        :type metadata: Sequence[Tuple[str, str]]
+        :param metadata: (Optional) Additional metadata that is provided to the method.
+
+        :type timeout: float or None
+        :param timeout:
+            (Optional) The amount of time, in seconds, to wait for the request to
+            complete.
 
+        :type retry: :class:`~google.api_core.retry_async.AsyncRetry` or :class:`~google.api_core.retry.Retry` or None
+        :param retry: (Optional) Designation of what errors, if any, should be retried.
         """
+        _validate_metadata(metadata)
         # The gRPC API requires the bucket name to be in the format "projects/_/buckets/bucket_name"
         bucket_path = f"projects/_/buckets/{bucket_name}"
         request = storage_v2.DeleteObjectRequest(
@@ -251,9 +277,21 @@ async def get_object(
         :param soft_deleted:
             (Optional) If True, return the soft-deleted version of this object.
 
+        :type metadata: Sequence[Tuple[str, str]]
+        :param metadata: (Optional) Additional metadata that is provided to the method.
+
+        :type timeout: float or None
+        :param timeout:
+            (Optional) The amount of time, in seconds, to wait for the request to
+            complete.
+
+        :type retry: :class:`~google.api_core.retry_async.AsyncRetry` or :class:`~google.api_core.retry.Retry` or None
+        :param retry: (Optional) Designation of what errors, if any, should be retried.
+
         :rtype: :class:`google.cloud._storage_v2.types.Object`
         :returns: The object metadata resource.
         """
+        _validate_metadata(metadata)
         bucket_path = f"projects/_/buckets/{bucket_name}"
 
         request = storage_v2.GetObjectRequest(

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

@@ -233,7 +233,6 @@ def __init__(
         self._open_retries: int = 0
         self.is_finalized: bool = False
         self.full_obj_server_crc32c: Optional[int] = None
-        self.metadata: Optional[List[Tuple[str, str]]] = None
 
     async def __aenter__(self):
         """Opens the underlying bidi-gRPC connection to read from the object."""
@@ -263,8 +262,6 @@ async def open(
         if self._is_stream_open:
             raise ValueError("Underlying bidi-gRPC stream is already open")
 
-        self.metadata = metadata
-
         if retry_policy is None:
 
             def on_error_wrapper(exc):
@@ -293,7 +290,7 @@ def combined_on_error(exc):
             )
 
         async def _do_open():
-            current_metadata = list(self.metadata) if self.metadata else []
+            current_metadata = list(metadata) if metadata else []
 
             # Cleanup stream from previous failed attempt, if any.
             if self.read_obj_str:
@@ -416,11 +413,6 @@ async def download_ranges(
 
         """
 
-        if metadata is not None:
-            self.metadata = metadata
-        else:
-            metadata = self.metadata
-
         if len(read_ranges) > 1000:
             raise ValueError(
                 "Invalid input - length of read_ranges cannot be more than 1000"

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

@@ -503,90 +503,3 @@ async def test_methods_require_open_stream_raises(self, mock_appendable_writer):
         for coro in methods:
             with pytest.raises(ValueError, match="Stream is not open"):
                 await coro
-
-    @pytest.mark.asyncio
-    async def test_append_persists_metadata_on_resumption(self, mock_appendable_writer):
-        # Arrange
-        mock_client = mock_appendable_writer["mock_client"]
-        mock_stream = mock_appendable_writer["mock_stream"]
-
-        test_metadata = [("custom-key", "custom-value")]
-        writer = self._make_one(mock_client)
-
-        # Act - Open with metadata
-        await writer.open(metadata=test_metadata)
-
-        # Assert first open used metadata
-        mock_stream.open.assert_called_once_with(metadata=test_metadata)
-        assert writer.metadata == test_metadata
-
-        # Setup resumption trigger
-        retryable_exc = exceptions.ServiceUnavailable("Retry me")
-        mock_stream.send.side_effect = retryable_exc
-
-        # Reset mock_stream.open call count to verify it is called again
-        mock_stream.open.reset_mock()
-
-        # Setup a fast retry policy to fail quickly in test
-        from google.api_core.retry_async import AsyncRetry
-
-        fast_retry = AsyncRetry(
-            predicate=lambda e: True,
-            initial=0.01,
-            maximum=0.01,
-            multiplier=1.0,
-            deadline=0.1,
-        )
-
-        # Act - append (should trigger retry and use stored metadata)
-        from google.api_core.exceptions import RetryError
-
-        try:
-            await writer.append(b"data", retry_policy=fast_retry)
-        except RetryError:
-            pass
-
-        # Assert second open (during retry) used the same test_metadata
-        mock_stream.open.assert_called_with(metadata=test_metadata)
-
-    @pytest.mark.asyncio
-    async def test_append_updates_and_persists_metadata_on_resumption(
-        self, mock_appendable_writer
-    ):
-        # Arrange
-        mock_client = mock_appendable_writer["mock_client"]
-        mock_stream = mock_appendable_writer["mock_stream"]
-
-        initial_metadata = [("initial-key", "initial-value")]
-        updated_metadata = [("updated-key", "updated-value")]
-        writer = self._make_one(mock_client)
-
-        # Act - Open with initial metadata
-        await writer.open(metadata=initial_metadata)
-        assert writer.metadata == initial_metadata
-
-        # Setup resumption trigger when append is called
-        retryable_exc = exceptions.ServiceUnavailable("Retry me")
-        mock_stream.send.side_effect = retryable_exc
-        mock_stream.open.reset_mock()
-
-        from google.api_core.retry_async import AsyncRetry
-        fast_retry = AsyncRetry(
-            predicate=lambda e: True,
-            initial=0.01,
-            maximum=0.01,
-            multiplier=1.0,
-            deadline=0.1,
-        )
-
-        from google.api_core.exceptions import RetryError
-        try:
-            await writer.append(
-                b"data", retry_policy=fast_retry, metadata=updated_metadata
-            )
-        except RetryError:
-            pass
-
-        # Assert writer.metadata was updated and used during stream reopening
-        assert writer.metadata == updated_metadata
-        mock_stream.open.assert_called_with(metadata=updated_metadata)

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

@@ -399,3 +399,35 @@ async def test_get_object_with_all_parameters(self, mock_async_storage_client):
         assert call_kwargs["metadata"] == metadata
         assert call_kwargs["timeout"] == timeout
         assert call_kwargs["retry"] == retry
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "invalid_metadata, expected_exc",
+        [
+            ("not-a-sequence", TypeError),
+            ([("key_only",)], ValueError),
+            ([("too", "many", "items")], ValueError),
+            ([(123, "val")], TypeError),
+            ([("key", 456)], TypeError),
+        ],
+    )
+    async def test_delete_object_invalid_metadata(self, invalid_metadata, expected_exc):
+        client = async_grpc_client.AsyncGrpcClient(credentials=_make_credentials())
+        with pytest.raises(expected_exc):
+            await client.delete_object("bucket", "object", metadata=invalid_metadata)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "invalid_metadata, expected_exc",
+        [
+            ("not-a-sequence", TypeError),
+            ([("key_only",)], ValueError),
+            ([("too", "many", "items")], ValueError),
+            ([(123, "val")], TypeError),
+            ([("key", 456)], TypeError),
+        ],
+    )
+    async def test_get_object_invalid_metadata(self, invalid_metadata, expected_exc):
+        client = async_grpc_client.AsyncGrpcClient(credentials=_make_credentials())
+        with pytest.raises(expected_exc):
+            await client.get_object("bucket", "object", metadata=invalid_metadata)