feat: add support for asynchronous rest streaming

.github/workflows/unittest.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        option: ["", "_grpc_gcp", "_wo_grpc", "_with_prerelease_deps"]
+        option: ["", "_grpc_gcp", "_wo_grpc", "_with_prerelease_deps", "_with_auth_aio"]
         python:
           - "3.7"
           - "3.8"

google/api_core/_rest_streaming_base.py

@@ -0,0 +1,118 @@
+# Copyright 2024 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.
+
+"""Helpers for server-side streaming in REST."""
+
+from collections import deque
+import string
+from typing import Deque, Union
+import types
+
+import proto
+import google.protobuf.message
+from google.protobuf.json_format import Parse
+
+
+class BaseResponseIterator:
+    """Base Iterator over REST API responses. This class should not be used directly.
+
+    Args:
+        response_message_cls (Union[proto.Message, google.protobuf.message.Message]): A response
+        class expected to be returned from an API.
+
+    Raises:
+        ValueError: If `response_message_cls` is not a subclass of `proto.Message` or `google.protobuf.message.Message`.
+    """
+
+    def __init__(
+        self,
+        response_message_cls: Union[proto.Message, google.protobuf.message.Message],
+    ):
+        self._response_message_cls = response_message_cls
+        # Contains a list of JSON responses ready to be sent to user.
+        self._ready_objs: Deque[str] = deque()
+        # Current JSON response being built.
+        self._obj = ""
+        # Keeps track of the nesting level within a JSON object.
+        self._level = 0
+        # Keeps track whether HTTP response is currently sending values
+        # inside of a string value.
+        self._in_string = False
+        # Whether an escape symbol "\" was encountered.
+        self._escape_next = False
+
+        self._grab = types.MethodType(self._create_grab(), self)
+
+    def _process_chunk(self, chunk: str):
+        if self._level == 0:
+            if chunk[0] != "[":
+                raise ValueError(
+                    "Can only parse array of JSON objects, instead got %s" % chunk
+                )
+        for char in chunk:
+            if char == "{":
+                if self._level == 1:
+                    # Level 1 corresponds to the outermost JSON object
+                    # (i.e. the one we care about).
+                    self._obj = ""
+                if not self._in_string:
+                    self._level += 1
+                self._obj += char
+            elif char == "}":
+                self._obj += char
+                if not self._in_string:
+                    self._level -= 1
+                if not self._in_string and self._level == 1:
+                    self._ready_objs.append(self._obj)
+            elif char == '"':
+                # Helps to deal with an escaped quotes inside of a string.
+                if not self._escape_next:
+                    self._in_string = not self._in_string
+                self._obj += char
+            elif char in string.whitespace:
+                if self._in_string:
+                    self._obj += char
+            elif char == "[":
+                if self._level == 0:
+                    self._level += 1
+                else:
+                    self._obj += char
+            elif char == "]":
+                if self._level == 1:
+                    self._level -= 1
+                else:
+                    self._obj += char
+            else:
+                self._obj += char
+            self._escape_next = not self._escape_next if char == "\\" else False
+
+    def _create_grab(self):
+        if issubclass(self._response_message_cls, proto.Message):
+
+            def grab(this):
+                return this._response_message_cls.from_json(
+                    this._ready_objs.popleft(), ignore_unknown_fields=True
+                )
+
+            return grab
+        elif issubclass(self._response_message_cls, google.protobuf.message.Message):
+
+            def grab(this):
+                return Parse(this._ready_objs.popleft(), this._response_message_cls())
+
+            return grab
+        else:
+            raise ValueError(
+                "Response message class must be a subclass of proto.Message or google.protobuf.message.Message."
+            )

google/api_core/rest_streaming.py

@@ -14,17 +14,15 @@
 
 """Helpers for server-side streaming in REST."""
 
-from collections import deque
-import string
-from typing import Deque, Union
+from typing import Union
 
 import proto
 import requests
 import google.protobuf.message
-from google.protobuf.json_format import Parse
+from google.api_core._rest_streaming_base import BaseResponseIterator
 
 
-class ResponseIterator:
+class ResponseIterator(BaseResponseIterator):
     """Iterator over REST API responses.
 
     Args:
@@ -33,7 +31,9 @@ class ResponseIterator:
         class expected to be returned from an API.
 
     Raises:
-        ValueError: If `response_message_cls` is not a subclass of `proto.Message` or `google.protobuf.message.Message`.
+        ValueError:
+            - If `response_message_cls` is not a subclass of `proto.Message` or `google.protobuf.message.Message`.
+            - If `response` is not an instance of `requests.Response`.
     """
 
     def __init__(
@@ -42,68 +42,16 @@ def __init__(
         response_message_cls: Union[proto.Message, google.protobuf.message.Message],
     ):
         self._response = response
-        self._response_message_cls = response_message_cls
         # Inner iterator over HTTP response's content.
         self._response_itr = self._response.iter_content(decode_unicode=True)
-        # Contains a list of JSON responses ready to be sent to user.
-        self._ready_objs: Deque[str] = deque()
-        # Current JSON response being built.
-        self._obj = ""
-        # Keeps track of the nesting level within a JSON object.
-        self._level = 0
-        # Keeps track whether HTTP response is currently sending values
-        # inside of a string value.
-        self._in_string = False
-        # Whether an escape symbol "\" was encountered.
-        self._escape_next = False
+        super(ResponseIterator, self).__init__(
+            response_message_cls=response_message_cls
+        )
 
     def cancel(self):
         """Cancel existing streaming operation."""
         self._response.close()
 
-    def _process_chunk(self, chunk: str):
-        if self._level == 0:
-            if chunk[0] != "[":
-                raise ValueError(
-                    "Can only parse array of JSON objects, instead got %s" % chunk
-                )
-        for char in chunk:
-            if char == "{":
-                if self._level == 1:
-                    # Level 1 corresponds to the outermost JSON object
-                    # (i.e. the one we care about).
-                    self._obj = ""
-                if not self._in_string:
-                    self._level += 1
-                self._obj += char
-            elif char == "}":
-                self._obj += char
-                if not self._in_string:
-                    self._level -= 1
-                if not self._in_string and self._level == 1:
-                    self._ready_objs.append(self._obj)
-            elif char == '"':
-                # Helps to deal with an escaped quotes inside of a string.
-                if not self._escape_next:
-                    self._in_string = not self._in_string
-                self._obj += char
-            elif char in string.whitespace:
-                if self._in_string:
-                    self._obj += char
-            elif char == "[":
-                if self._level == 0:
-                    self._level += 1
-                else:
-                    self._obj += char
-            elif char == "]":
-                if self._level == 1:
-                    self._level -= 1
-                else:
-                    self._obj += char
-            else:
-                self._obj += char
-            self._escape_next = not self._escape_next if char == "\\" else False
-
     def __next__(self):
         while not self._ready_objs:
             try:
@@ -115,18 +63,5 @@ def __next__(self):
                 raise e
         return self._grab()
 
-    def _grab(self):
-        # Add extra quotes to make json.loads happy.
-        if issubclass(self._response_message_cls, proto.Message):
-            return self._response_message_cls.from_json(
-                self._ready_objs.popleft(), ignore_unknown_fields=True
-            )
-        elif issubclass(self._response_message_cls, google.protobuf.message.Message):
-            return Parse(self._ready_objs.popleft(), self._response_message_cls())
-        else:
-            raise ValueError(
-                "Response message class must be a subclass of proto.Message or google.protobuf.message.Message."
-            )
-
     def __iter__(self):
         return self

google/api_core/rest_streaming_async.py

@@ -0,0 +1,85 @@
+# Copyright 2024 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.
+
+"""Helpers for asynchronous server-side streaming in REST."""
+
+from typing import Union
+
+import proto
+
+try:
+    import google.auth.aio.transport
+except ImportError:  # pragma: NO COVER
+    # TODO (ohmayr): Update this version once auth work is released.
+    raise ValueError(
+        "google-auth>=2.3x.x is required to use asynchronous rest streaming."
+    )
+
+import google.protobuf.message
+from google.api_core._rest_streaming_base import BaseResponseIterator
+
+
+class AsyncResponseIterator(BaseResponseIterator):
+    """Asynchronous Iterator over REST API responses.
+
+    Args:
+        response (google.auth.aio.transport.Response): An API response object.
+        response_message_cls (Union[proto.Message, google.protobuf.message.Message]): A response
+        class expected to be returned from an API.
+
+    Raises:
+        ValueError:
+            - If `response_message_cls` is not a subclass of `proto.Message` or `google.protobuf.message.Message`.
+            - If `response` is not an instance of a subclass of `google.auth.aio.transport.Response`.
+    """
+
+    def __init__(
+        self,
+        response: google.auth.aio.transport.Response,
+        response_message_cls: Union[proto.Message, google.protobuf.message.Message],
+    ):
+        self._response = response
+        self._chunk_size = 1024
+        self._response_itr = self._response.content().__aiter__()
+        super(AsyncResponseIterator, self).__init__(
+            response_message_cls=response_message_cls
+        )
+
+    async def __aenter__(self):
+        return self
+
+    async def cancel(self):
+        """Cancel existing streaming operation."""
+        await self._response.close()
+
+    async def __anext__(self):
+        while not self._ready_objs:
+            try:
+                chunk = await self._response_itr.__anext__()
+                chunk = chunk.decode("utf-8")
+                self._process_chunk(chunk)
+            except StopAsyncIteration as e:
+                if self._level > 0:
+                    raise ValueError("i Unfinished stream: %s" % self._obj)
+                raise e
+            except ValueError as e:
+                raise e
+        return self._grab()
+
+    def __aiter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        """Cancel existing async streaming operation."""
+        await self._response.close()

noxfile.py

@@ -38,6 +38,7 @@
     "unit",
     "unit_grpc_gcp",
     "unit_wo_grpc",
+    "unit_with_auth_aio",
     "cover",
     "pytype",
     "mypy",
@@ -109,7 +110,7 @@ def install_prerelease_dependencies(session, constraints_path):
         session.install(*other_deps)
 
 
-def default(session, install_grpc=True, prerelease=False):
+def default(session, install_grpc=True, prerelease=False, install_auth_aio=False):
     """Default unit test session.
 
     This is intended to be run **without** an interpreter set, so
@@ -144,6 +145,11 @@ def default(session, install_grpc=True, prerelease=False):
             f"{constraints_dir}/constraints-{session.python}.txt",
         )
 
+    if install_auth_aio:
+        session.install(
+            "google-auth @ git+https://git@github.com/googleapis/google-auth-library-python@8833ad6f92c3300d6645355994c7db2356bd30ad"
+        )
+
     # Print out package versions of dependencies
     session.run(
         "python", "-c", "import google.protobuf; print(google.protobuf.__version__)"
@@ -229,6 +235,12 @@ def unit_wo_grpc(session):
     default(session, install_grpc=False)
 
 
+@nox.session(python=PYTHON_VERSIONS)
+def unit_with_auth_aio(session):
+    """Run the unit test suite with google.auth.aio installed"""
+    default(session, install_auth_aio=True)
+
+
 @nox.session(python=DEFAULT_PYTHON_VERSION)
 def lint_setup_py(session):
     """Verify that setup.py is valid (including RST check)."""