From 21e66a3557e87760bcd291cdcc5511cd3799457e Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Wed, 17 Sep 2025 09:59:21 +0000 Subject: [PATCH 1/9] add AsyncAbstractObjectStream this will be the parent class for AsyncReadObjectStream and AsyncWriteObjectStream --- .../asyncio/async_abstract_object_stream.py | 36 +++++++++++ .../test_async_abstract_object_stream.py | 64 +++++++++++++++++++ 2 files changed, 100 insertions(+) create mode 100644 google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py create mode 100644 tests/unit/asyncio/test_async_abstract_object_stream.py diff --git a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py new file mode 100644 index 000000000..02e72ffae --- /dev/null +++ b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py @@ -0,0 +1,36 @@ +import abc + + +class AsyncAbstractObjectStream(abc.ABC): + """ + Abstract class for both ReadObjectStream as well as WriteObjectStream. + + Attributes will include + 1. bucket_name + 2. object_name + 3. generation_number (if given) + + + """ + + def __init__(self, bucket_name, object_name, generation_number=None): + super().__init__() + self.bucket_name = bucket_name + self.object_name = object_name + self.generation_number = generation_number + + @abc.abstractmethod + async def open(self): + raise NotImplementedError("Subclasses should implement this method.") + + @abc.abstractmethod + async def close(self): + raise NotImplementedError("Subclasses should implement this method.") + + @abc.abstractmethod + async def send(self): + raise NotImplementedError("Subclasses should implement this method.") + + @abc.abstractmethod + async def recv(self): + raise NotImplementedError("Subclasses should implement this method.") diff --git a/tests/unit/asyncio/test_async_abstract_object_stream.py b/tests/unit/asyncio/test_async_abstract_object_stream.py new file mode 100644 index 000000000..9679d729e --- /dev/null +++ b/tests/unit/asyncio/test_async_abstract_object_stream.py @@ -0,0 +1,64 @@ +# Copyright 2025 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 pytest + +from google.cloud.storage._experimental.asyncio.async_abstract_object_stream import ( + AsyncAbstractObjectStream, +) + + +# A concrete implementation for testing purposes. +class _ConcreteStream(AsyncAbstractObjectStream): + async def open(self): + pass + + async def close(self): + pass + + async def send(self): + pass + + async def recv(self): + pass + + +def test_init(): + """Test the constructor of AsyncAbstractObjectStream.""" + bucket_name = "test-bucket" + object_name = "test-object" + generation = 12345 + + # Test with all parameters + stream = _ConcreteStream(bucket_name, object_name, generation_number=generation) + assert stream.bucket_name == bucket_name + assert stream.object_name == object_name + assert stream.generation_number == generation + + # Test with default generation_number + stream_no_gen = _ConcreteStream(bucket_name, object_name) + assert stream_no_gen.bucket_name == bucket_name + assert stream_no_gen.object_name == object_name + assert stream_no_gen.generation_number is None + + +def test_instantiation_fails_without_implementation(): + """Test that instantiating an incomplete subclass raises TypeError.""" + + class _IncompleteStream(AsyncAbstractObjectStream): + # Missing implementations for abstract methods like open(), close(), etc. + pass + + with pytest.raises(TypeError, match="Can't instantiate abstract class"): + _IncompleteStream("bucket", "object") From 39503f49553daa611cfe0a3425fb487b502dca29 Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Wed, 17 Sep 2025 10:22:06 +0000 Subject: [PATCH 2/9] keep _AsyncAbstractObjectStream private --- .../_experimental/asyncio/async_abstract_object_stream.py | 2 +- tests/unit/asyncio/test_async_abstract_object_stream.py | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py index 02e72ffae..86de7f715 100644 --- a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py @@ -1,7 +1,7 @@ import abc -class AsyncAbstractObjectStream(abc.ABC): +class _AsyncAbstractObjectStream(abc.ABC): """ Abstract class for both ReadObjectStream as well as WriteObjectStream. diff --git a/tests/unit/asyncio/test_async_abstract_object_stream.py b/tests/unit/asyncio/test_async_abstract_object_stream.py index 9679d729e..e0dc130ea 100644 --- a/tests/unit/asyncio/test_async_abstract_object_stream.py +++ b/tests/unit/asyncio/test_async_abstract_object_stream.py @@ -15,12 +15,12 @@ import pytest from google.cloud.storage._experimental.asyncio.async_abstract_object_stream import ( - AsyncAbstractObjectStream, + _AsyncAbstractObjectStream, ) # A concrete implementation for testing purposes. -class _ConcreteStream(AsyncAbstractObjectStream): +class _ConcreteStream(_AsyncAbstractObjectStream): async def open(self): pass @@ -56,7 +56,7 @@ def test_init(): def test_instantiation_fails_without_implementation(): """Test that instantiating an incomplete subclass raises TypeError.""" - class _IncompleteStream(AsyncAbstractObjectStream): + class _IncompleteStream(_AsyncAbstractObjectStream): # Missing implementations for abstract methods like open(), close(), etc. pass From 0810afc315e29a4f9caa25c08faa7840281de545 Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Thu, 18 Sep 2025 11:00:17 +0000 Subject: [PATCH 3/9] fix doc strings, add licence and type hints --- .../asyncio/async_abstract_object_stream.py | 53 ++++++++++++++----- 1 file changed, 39 insertions(+), 14 deletions(-) diff --git a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py index 86de7f715..03e2c5690 100644 --- a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py @@ -1,36 +1,61 @@ +# Copyright 2025 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 abc +from typing import Any, Optional class _AsyncAbstractObjectStream(abc.ABC): - """ - Abstract class for both ReadObjectStream as well as WriteObjectStream. + """Abstract base class for asynchronous object streams. + + This class defines the common interface for both reading from and writing + to a GCS object in a streaming fashion. - Attributes will include - 1. bucket_name - 2. object_name - 3. generation_number (if given) + :type bucket_name: str + :param bucket_name: (Optional) The name of the bucket containing the object. + :type object_name: str + :param object_name: (Optional) The name of the object. + :type generation_number: int + :param generation_number: (Optional) If present, selects a specific revision of + this object. """ - def __init__(self, bucket_name, object_name, generation_number=None): + def __init__( + self, + bucket_name: Optional[str] = None, + object_name: Optional[str] = None, + generation_number: Optional[int] = None, + ) -> None: super().__init__() - self.bucket_name = bucket_name - self.object_name = object_name - self.generation_number = generation_number + self.bucket_name: Optional[str] = bucket_name + self.object_name: Optional[str] = object_name + self.generation_number: Optional[int] = generation_number @abc.abstractmethod - async def open(self): + async def open(self) -> None: raise NotImplementedError("Subclasses should implement this method.") @abc.abstractmethod - async def close(self): + async def close(self) -> None: raise NotImplementedError("Subclasses should implement this method.") @abc.abstractmethod - async def send(self): + async def send(self, message: Any) -> None: raise NotImplementedError("Subclasses should implement this method.") @abc.abstractmethod - async def recv(self): + async def recv(self) -> Any: raise NotImplementedError("Subclasses should implement this method.") From a14bc6895ad704beb57438b55ede8f6487b50f9e Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Thu, 18 Sep 2025 11:16:30 +0000 Subject: [PATCH 4/9] pass abstract methods --- .../_experimental/asyncio/async_abstract_object_stream.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py index 03e2c5690..752a058c1 100644 --- a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py @@ -46,16 +46,16 @@ def __init__( @abc.abstractmethod async def open(self) -> None: - raise NotImplementedError("Subclasses should implement this method.") + pass @abc.abstractmethod async def close(self) -> None: - raise NotImplementedError("Subclasses should implement this method.") + pass @abc.abstractmethod async def send(self, message: Any) -> None: - raise NotImplementedError("Subclasses should implement this method.") + pass @abc.abstractmethod async def recv(self) -> Any: - raise NotImplementedError("Subclasses should implement this method.") + pass From 635ad07ea962f0bb0df491ac57631d167be29369 Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Thu, 18 Sep 2025 11:29:33 +0000 Subject: [PATCH 5/9] add handle param --- .../_experimental/asyncio/async_abstract_object_stream.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py index 752a058c1..28a92ccb4 100644 --- a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py @@ -31,6 +31,10 @@ class _AsyncAbstractObjectStream(abc.ABC): :type generation_number: int :param generation_number: (Optional) If present, selects a specific revision of this object. + + :type handle: bytes + :param handle: (Optional) The handle for the object, could be read_handle or + write_handle, based on how the stream is used. """ def __init__( @@ -38,11 +42,13 @@ def __init__( bucket_name: Optional[str] = None, object_name: Optional[str] = None, generation_number: Optional[int] = None, + handle: Optional[bytes] = None, ) -> None: super().__init__() self.bucket_name: Optional[str] = bucket_name self.object_name: Optional[str] = object_name self.generation_number: Optional[int] = generation_number + self.handle: Optional[bytes] = handle @abc.abstractmethod async def open(self) -> None: From ba453d4617a6c197078e9a3783730d48d5e3b1a3 Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Thu, 18 Sep 2025 11:36:07 +0000 Subject: [PATCH 6/9] include handle in tests --- tests/unit/asyncio/test_async_abstract_object_stream.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/unit/asyncio/test_async_abstract_object_stream.py b/tests/unit/asyncio/test_async_abstract_object_stream.py index e0dc130ea..d81c9ef56 100644 --- a/tests/unit/asyncio/test_async_abstract_object_stream.py +++ b/tests/unit/asyncio/test_async_abstract_object_stream.py @@ -39,18 +39,23 @@ def test_init(): bucket_name = "test-bucket" object_name = "test-object" generation = 12345 + handle = b"test-handle" # Test with all parameters - stream = _ConcreteStream(bucket_name, object_name, generation_number=generation) + stream = _ConcreteStream( + bucket_name, object_name, generation_number=generation, handle=handle + ) assert stream.bucket_name == bucket_name assert stream.object_name == object_name assert stream.generation_number == generation + assert stream.handle == handle # Test with default generation_number stream_no_gen = _ConcreteStream(bucket_name, object_name) assert stream_no_gen.bucket_name == bucket_name assert stream_no_gen.object_name == object_name assert stream_no_gen.generation_number is None + assert stream_no_gen.handle is None def test_instantiation_fails_without_implementation(): From 800c6df51e19123b022b9aa336cda59f939444bf Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Thu, 18 Sep 2025 14:59:47 +0000 Subject: [PATCH 7/9] remove unit tests for abstract class --- .../asyncio/async_abstract_object_stream.py | 8 +-- .../test_async_abstract_object_stream.py | 69 ------------------- 2 files changed, 4 insertions(+), 73 deletions(-) delete mode 100644 tests/unit/asyncio/test_async_abstract_object_stream.py diff --git a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py index 28a92ccb4..325089ba5 100644 --- a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py @@ -17,10 +17,10 @@ class _AsyncAbstractObjectStream(abc.ABC): - """Abstract base class for asynchronous object streams. + """Abstract base class to represent gRPC stream for GCS ``Object``. - This class defines the common interface for both reading from and writing - to a GCS object in a streaming fashion. + Concrete implementation of this class could be ``_AsyncReadObjectStream`` + or ``_AsyncWriteObjectStream``. :type bucket_name: str :param bucket_name: (Optional) The name of the bucket containing the object. @@ -59,7 +59,7 @@ async def close(self) -> None: pass @abc.abstractmethod - async def send(self, message: Any) -> None: + async def send(self, protobuf: Any) -> None: pass @abc.abstractmethod diff --git a/tests/unit/asyncio/test_async_abstract_object_stream.py b/tests/unit/asyncio/test_async_abstract_object_stream.py deleted file mode 100644 index d81c9ef56..000000000 --- a/tests/unit/asyncio/test_async_abstract_object_stream.py +++ /dev/null @@ -1,69 +0,0 @@ -# Copyright 2025 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 pytest - -from google.cloud.storage._experimental.asyncio.async_abstract_object_stream import ( - _AsyncAbstractObjectStream, -) - - -# A concrete implementation for testing purposes. -class _ConcreteStream(_AsyncAbstractObjectStream): - async def open(self): - pass - - async def close(self): - pass - - async def send(self): - pass - - async def recv(self): - pass - - -def test_init(): - """Test the constructor of AsyncAbstractObjectStream.""" - bucket_name = "test-bucket" - object_name = "test-object" - generation = 12345 - handle = b"test-handle" - - # Test with all parameters - stream = _ConcreteStream( - bucket_name, object_name, generation_number=generation, handle=handle - ) - assert stream.bucket_name == bucket_name - assert stream.object_name == object_name - assert stream.generation_number == generation - assert stream.handle == handle - - # Test with default generation_number - stream_no_gen = _ConcreteStream(bucket_name, object_name) - assert stream_no_gen.bucket_name == bucket_name - assert stream_no_gen.object_name == object_name - assert stream_no_gen.generation_number is None - assert stream_no_gen.handle is None - - -def test_instantiation_fails_without_implementation(): - """Test that instantiating an incomplete subclass raises TypeError.""" - - class _IncompleteStream(_AsyncAbstractObjectStream): - # Missing implementations for abstract methods like open(), close(), etc. - pass - - with pytest.raises(TypeError, match="Can't instantiate abstract class"): - _IncompleteStream("bucket", "object") From 6dec6c692779e26ddcf30588375f4d0260a7102d Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Thu, 18 Sep 2025 18:31:13 +0000 Subject: [PATCH 8/9] bucket_name and object_name cannot be NONE --- .../_experimental/asyncio/async_abstract_object_stream.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py index 325089ba5..1ba5aef9b 100644 --- a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py @@ -39,14 +39,14 @@ class _AsyncAbstractObjectStream(abc.ABC): def __init__( self, - bucket_name: Optional[str] = None, - object_name: Optional[str] = None, + bucket_name: str, + object_name: str, generation_number: Optional[int] = None, handle: Optional[bytes] = None, ) -> None: super().__init__() - self.bucket_name: Optional[str] = bucket_name - self.object_name: Optional[str] = object_name + self.bucket_name: str = bucket_name + self.object_name: str = object_name self.generation_number: Optional[int] = generation_number self.handle: Optional[bytes] = handle From 2054989c7b83ba86d2461e046c972d61e1cbbe49 Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Fri, 19 Sep 2025 06:26:06 +0000 Subject: [PATCH 9/9] minor edit - add bidi-stream in doc string --- .../_experimental/asyncio/async_abstract_object_stream.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py index 1ba5aef9b..49d7a293a 100644 --- a/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_abstract_object_stream.py @@ -17,7 +17,7 @@ class _AsyncAbstractObjectStream(abc.ABC): - """Abstract base class to represent gRPC stream for GCS ``Object``. + """Abstract base class to represent gRPC bidi-stream for GCS ``Object``. Concrete implementation of this class could be ``_AsyncReadObjectStream`` or ``_AsyncWriteObjectStream``.