feat: Implement asynchronous AuthorizedSession api response class (#1575)

* feat: implement asynchronous response class for AuthorizedSessions API

* check if aiohttp is installed and avoid tests dependency

* update content to be async

* update docstring to be specific to aiohttp

* add type checking and avoid leaking underlying API responses

* add test case for iterating chunks

* add read method to response interface

* address PR comments

* fix lint issues

Author: ohmayr

google/auth/aio/transport/__init__.py

@@ -23,3 +23,59 @@
 :mod:`google.auth` to make asynchronous requests. :class:`Response` defines the interface
 for the return value of :class:`Request`.
 """
+
+import abc
+from typing import AsyncGenerator, Dict
+
+
+class Response(metaclass=abc.ABCMeta):
+    """Asynchronous HTTP Response Interface."""
+
+    @property
+    @abc.abstractmethod
+    def status_code(self) -> int:
+        """
+        The HTTP response status code..
+
+        Returns:
+            int: The HTTP response status code.
+
+        """
+        raise NotImplementedError("status_code must be implemented.")
+
+    @property
+    @abc.abstractmethod
+    def headers(self) -> Dict[str, str]:
+        """The HTTP response headers.
+
+        Returns:
+            Dict[str, str]: The HTTP response headers.
+        """
+        raise NotImplementedError("headers must be implemented.")
+
+    @abc.abstractmethod
+    async def content(self, chunk_size: int = 1024) -> AsyncGenerator[bytes, None]:
+        """The raw response content.
+
+        Args:
+            chunk_size (int): The size of each chunk. Defaults to 1024.
+
+        Yields:
+            AsyncGenerator[bytes, None]: An asynchronous generator yielding
+            response chunks as bytes.
+        """
+        raise NotImplementedError("content must be implemented.")
+
+    @abc.abstractmethod
+    async def read(self) -> bytes:
+        """Read the entire response content as bytes.
+
+        Returns:
+            bytes: The entire response content.
+        """
+        raise NotImplementedError("read must be implemented.")
+
+    @abc.abstractmethod
+    async def close(self):
+        """Close the response after it is fully consumed to resource."""
+        raise NotImplementedError("close must be implemented.")

google/auth/aio/transport/aiohttp.py

@@ -12,10 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Transport adapter for Asynchronous HTTP Requests.
+"""Transport adapter for AIOHTTP Requests.
 """
 
+try:
+    import aiohttp
+except ImportError as caught_exc:  # pragma: NO COVER
+    raise ImportError(
+        "The aiohttp library is not installed from please install the aiohttp package to use the aiohttp transport."
+    ) from caught_exc
+from typing import AsyncGenerator, Dict
 
+from google.auth import _helpers
+from google.auth.aio import transport
 from google.auth.exceptions import TimeoutError
 
 import asyncio
@@ -65,3 +74,43 @@ async def with_timeout(coro):
 
     finally:
         _remaining_time()
+
+
+class Response(transport.Response):
+
+    """
+    Represents an HTTP response and its data. It is returned by ``google.auth.aio.transport.sessions.AuthorizedSession``.
+
+    Args:
+        response (aiohttp.ClientResponse): An instance of aiohttp.ClientResponse.
+
+    Attributes:
+        status_code (int): The HTTP status code of the response.
+        headers (Dict[str, str]): A case-insensitive multidict proxy wiht HTTP headers of response.
+    """
+
+    def __init__(self, response: aiohttp.ClientResponse):
+        self._response = response
+
+    @property
+    @_helpers.copy_docstring(transport.Response)
+    def status_code(self) -> int:
+        return self._response.status
+
+    @property
+    @_helpers.copy_docstring(transport.Response)
+    def headers(self) -> Dict[str, str]:
+        return {key: value for key, value in self._response.headers.items()}
+
+    @_helpers.copy_docstring(transport.Response)
+    async def content(self, chunk_size: int = 1024) -> AsyncGenerator[bytes, None]:
+        async for chunk in self._response.content.iter_chunked(chunk_size):
+            yield chunk
+
+    @_helpers.copy_docstring(transport.Response)
+    async def read(self) -> bytes:
+        return await self._response.read()
+
+    @_helpers.copy_docstring(transport.Response)
+    async def close(self):
+        return await self._response.close()

tests/transport/aio/test_aiohttp.py

@@ -12,18 +12,76 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import google.auth.aio.transport.aiohttp as auth_aiohttp
+from unittest.mock import AsyncMock, Mock, patch
+
 import pytest  # type: ignore
+
 import asyncio
+
+import google.auth.aio.transport.aiohttp as auth_aiohttp
+
 from google.auth.exceptions import TimeoutError
-from unittest.mock import patch
 
 
 @pytest.fixture
 async def simple_async_task():
     return True
 
 
+@pytest.fixture
+def mock_response():
+    response = Mock()
+    response.status = 200
+    response.headers = {"Content-Type": "application/json", "Content-Length": "100"}
+    mock_iterator = AsyncMock()
+    mock_iterator.__aiter__.return_value = iter(
+        [b"Cavefish ", b"have ", b"no ", b"sight."]
+    )
+    response.content.iter_chunked = lambda chunk_size: mock_iterator
+    response.read = AsyncMock(return_value=b"Cavefish have no sight.")
+    response.close = AsyncMock()
+
+    return auth_aiohttp.Response(response)
+
+
+class TestResponse(object):
+    @pytest.mark.asyncio
+    async def test_response_status_code(self, mock_response):
+        assert mock_response.status_code == 200
+
+    @pytest.mark.asyncio
+    async def test_response_headers(self, mock_response):
+        assert mock_response.headers["Content-Type"] == "application/json"
+        assert mock_response.headers["Content-Length"] == "100"
+
+    @pytest.mark.asyncio
+    async def test_response_content(self, mock_response):
+        content = b"".join([chunk async for chunk in mock_response.content()])
+        assert content == b"Cavefish have no sight."
+
+    @pytest.mark.asyncio
+    async def test_response_read(self, mock_response):
+        content = await mock_response.read()
+        assert content == b"Cavefish have no sight."
+
+    @pytest.mark.asyncio
+    async def test_response_close(self, mock_response):
+        await mock_response.close()
+        mock_response._response.close.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_response_content_stream(self, mock_response):
+        itr = mock_response.content().__aiter__()
+        content = []
+        try:
+            while True:
+                chunk = await itr.__anext__()
+                content.append(chunk)
+        except StopAsyncIteration:
+            pass
+        assert b"".join(content) == b"Cavefish have no sight."
+
+
 class TestTimeoutGuard(object):
     default_timeout = 1