[rest] switch base responses to ABCs

sdk/core/azure-core/CHANGELOG.md

@@ -10,6 +10,8 @@
 
 - The `text` property on `azure.core.rest.HttpResponse` and `azure.core.rest.AsyncHttpResponse` has changed to a method, which also takes
 an `encoding` parameter.
+- `azure.core.rest.HttpResponse` and `azure.core.rest.AsyncHttpResponse` are now abstract base classes. They should not be initialized directly, instead
+your transport responses should inherit from them and implement them.
 
 ### Bugs Fixed
 

sdk/core/azure-core/azure/core/_pipeline_client_async.py

@@ -25,7 +25,7 @@
 # --------------------------------------------------------------------------
 
 import logging
-from collections.abc import Iterable
+import collections.abc
 from typing import Any, Awaitable
 from .configuration import Configuration
 from .pipeline import AsyncPipeline
@@ -63,6 +63,26 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+class _AsyncContextManager(collections.abc.Awaitable):
+
+    def __init__(self, wrapped: collections.abc.Awaitable):
+        super().__init__()
+        self.wrapped = wrapped
+        self.response = None
+
+    def __await__(self):
+        return self.wrapped.__await__()
+
+    async def __aenter__(self):
+        self.response = await self
+        return self.response
+
+    async def __aexit__(self, *args):
+        await self.response.__aexit__(*args)
+
+    async def close(self):
+        await self.response.close()
+
 
 class AsyncPipelineClient(PipelineClientBase):
     """Service client core methods.
@@ -126,7 +146,7 @@ def _build_pipeline(self, config, **kwargs): # pylint: disable=no-self-use
                 config.proxy_policy,
                 ContentDecodePolicy(**kwargs)
             ]
-            if isinstance(per_call_policies, Iterable):
+            if isinstance(per_call_policies, collections.abc.Iterable):
                 policies.extend(per_call_policies)
             else:
                 policies.append(per_call_policies)
@@ -135,7 +155,7 @@ def _build_pipeline(self, config, **kwargs): # pylint: disable=no-self-use
                              config.retry_policy,
                              config.authentication_policy,
                              config.custom_hook_policy])
-            if isinstance(per_retry_policies, Iterable):
+            if isinstance(per_retry_policies, collections.abc.Iterable):
                 policies.extend(per_retry_policies)
             else:
                 policies.append(per_retry_policies)
@@ -144,13 +164,13 @@ def _build_pipeline(self, config, **kwargs): # pylint: disable=no-self-use
                              DistributedTracingPolicy(**kwargs),
                              config.http_logging_policy or HttpLoggingPolicy(**kwargs)])
         else:
-            if isinstance(per_call_policies, Iterable):
+            if isinstance(per_call_policies, collections.abc.Iterable):
                 per_call_policies_list = list(per_call_policies)
             else:
                 per_call_policies_list = [per_call_policies]
             per_call_policies_list.extend(policies)
             policies = per_call_policies_list
-            if isinstance(per_retry_policies, Iterable):
+            if isinstance(per_retry_policies, collections.abc.Iterable):
                 per_retry_policies_list = list(per_retry_policies)
             else:
                 per_retry_policies_list = [per_retry_policies]
@@ -189,6 +209,7 @@ async def _make_pipeline_call(self, request, **kwargs):
                     # the body is loaded. instead of doing response.read(), going to set the body
                     # to the internal content
                     rest_response._content = response.body()  # pylint: disable=protected-access
+                    await rest_response.read()
                     await rest_response.close()
                 except Exception as exc:
                     await rest_response.close()
@@ -223,6 +244,5 @@ def send_request(
         :return: The response of your network call. Does not do error handling on your response.
         :rtype: ~azure.core.rest.AsyncHttpResponse
         """
-        from .rest._rest_py3 import _AsyncContextManager
         wrapped = self._make_pipeline_call(request, stream=stream, **kwargs)
         return _AsyncContextManager(wrapped=wrapped)

sdk/core/azure-core/azure/core/pipeline/_tools.py

@@ -46,11 +46,10 @@ def to_rest_request(pipeline_transport_request):
 def to_rest_response(pipeline_transport_response):
     from .transport._requests_basic import RequestsTransportResponse
     from ..rest._requests_basic import RestRequestsTransportResponse
-    from ..rest import HttpResponse
     if isinstance(pipeline_transport_response, RequestsTransportResponse):
         response_type = RestRequestsTransportResponse
     else:
-        response_type = HttpResponse
+        raise ValueError("Unknown transport response")
     response = response_type(
         request=to_rest_request(pipeline_transport_response.request),
         internal_response=pipeline_transport_response.internal_response,

sdk/core/azure-core/azure/core/pipeline/_tools_async.py

@@ -55,8 +55,7 @@ def _get_response_type(pipeline_transport_response):
             return RestTrioRequestsTransportResponse
     except ImportError:
         pass
-    from ..rest import AsyncHttpResponse
-    return AsyncHttpResponse
+    raise ValueError("Unknown transport response")
 
 def to_rest_response(pipeline_transport_response):
     response_type = _get_response_type(pipeline_transport_response)

sdk/core/azure-core/azure/core/rest/_aiohttp.py

@@ -27,23 +27,42 @@
 import asyncio
 from typing import AsyncIterator
 from multidict import CIMultiDict
-from . import HttpRequest, AsyncHttpResponse
+from . import HttpRequest
+from ._http_response_impl_async import AsyncHttpResponseImpl
+from ._helpers import HeadersType
 from ._helpers_py3 import iter_raw_helper, iter_bytes_helper
 from ..pipeline.transport._aiohttp import AioHttpStreamDownloadGenerator
 
 
-class RestAioHttpTransportResponse(AsyncHttpResponse):
+class RestAioHttpTransportResponse(AsyncHttpResponseImpl):
     def __init__(
         self,
         *,
         request: HttpRequest,
         internal_response,
     ):
         super().__init__(request=request, internal_response=internal_response)
-        self.status_code = internal_response.status
-        self.headers = CIMultiDict(internal_response.headers)  # type: ignore
-        self.reason = internal_response.reason
-        self.content_type = internal_response.headers.get('content-type')
+        self._headers = CIMultiDict(internal_response.headers)  # type: ignore
+
+    @property
+    def status_code(self) -> int:
+        """The status code of this response"""
+        return self._internal_response.status
+
+    @property
+    def headers(self) -> HeadersType:
+        """The response headers"""
+        return self._headers
+
+    @property
+    def content_type(self) -> str:
+        """The content type of the response"""
+        return self._internal_response.headers.get('content-type')
+
+    @property
+    def reason(self) -> str:
+        """The reason phrase for this response"""
+        return self._internal_response.reason
 
     async def iter_raw(self) -> AsyncIterator[bytes]:
         """Asynchronously iterates over the response's bytes. Will not decompress in the process
@@ -82,6 +101,6 @@ async def close(self) -> None:
         :return: None
         :rtype: None
         """
-        self.is_closed = True
+        self._is_closed = True
         self._internal_response.close()
         await asyncio.sleep(0)

sdk/core/azure-core/azure/core/rest/_helpers_py3.py

@@ -65,7 +65,7 @@ def _stream_download_helper(
     if response.is_closed:
         raise StreamClosedError(response)
 
-    response.is_stream_consumed = True
+    response._is_stream_consumed = True  # pylint: disable=protected-access
     return stream_download_generator(
         pipeline=None,
         response=response,

sdk/core/azure-core/azure/core/rest/_http_response_impl.py

@@ -0,0 +1,208 @@
+# --------------------------------------------------------------------------
+#
+# Copyright (c) Microsoft Corporation. All rights reserved.
+#
+# The MIT License (MIT)
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the ""Software""), to
+# deal in the Software without restriction, including without limitation the
+# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+# sell copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+# IN THE SOFTWARE.
+#
+# --------------------------------------------------------------------------
+from json import loads
+from typing import cast, Any, Optional, Iterator
+from ._helpers import (
+    get_charset_encoding,
+    decode_to_text,
+    parse_lines_from_text,
+)
+from ..exceptions import HttpResponseError, ResponseNotReadError
+try:
+    from ._rest_py3 import (
+        _HttpResponseBase,
+        HttpResponse as _HttpResponse,
+        HttpRequest as _HttpRequest
+    )
+except (SyntaxError, ImportError):
+    from ._rest import (  # type: ignore
+        _HttpResponseBase,
+        HttpResponse as _HttpResponse,
+        HttpRequest as _HttpRequest
+    )
+
+
+class _HttpResponseBaseImpl(_HttpResponseBase):
+
+    def __init__(self, **kwargs):
+        # type: (Any) -> None
+        super(_HttpResponseBaseImpl, self).__init__()
+        self._request = kwargs.pop("request")
+        self._internal_response = kwargs.pop("internal_response")
+        self._is_closed = False
+        self._is_stream_consumed = False
+        self._connection_data_block_size = None
+        self._json = None  # this is filled in ContentDecodePolicy, when we deserialize
+        self._content = None  # type: Optional[bytes]
+        self._text = None  # type: Optional[str]
+
+    @property
+    def request(self):
+        # type: (...) -> _HttpRequest
+        return self._request
+
+    @property
+    def url(self):
+        # type: (...) -> str
+        """Returns the URL that resulted in this response"""
+        return self.request.url
+
+    @property
+    def is_closed(self):
+        # type: (...) -> bool
+        """Whether the network connection has been closed yet"""
+        return self._is_closed
+
+    @property
+    def is_stream_consumed(self):
+        # type: (...) -> bool
+        """Whether the stream has been fully consumed"""
+        return self._is_stream_consumed
+
+    @property
+    def encoding(self):
+        # type: (...) -> Optional[str]
+        """Returns the response encoding.
+
+        :return: The response encoding. We either return the encoding set by the user,
+         or try extracting the encoding from the response's content type. If all fails,
+         we return `None`.
+        :rtype: optional[str]
+        """
+        try:
+            return self._encoding
+        except AttributeError:
+            self._encoding = get_charset_encoding(self)  # type: Optional[str]
+            return self._encoding
+
+    @encoding.setter
+    def encoding(self, value):
+        # type: (str) -> None
+        """Sets the response encoding"""
+        self._encoding = value
+        self._text = None  # clear text cache
+
+    def text(self, encoding=None):
+        # type: (Optional[str]) -> str
+        """Returns the response body as a string
+
+        :param optional[str] encoding: The encoding you want to decode the text with. Can
+         also be set independently through our encoding property
+        :return: The response's content decoded as a string.
+        """
+        if self._text is None or encoding:
+            encoding_to_pass = encoding or self.encoding
+            self._text = decode_to_text(encoding_to_pass, self.content)
+        return self._text
+
+    def json(self):
+        # type: (...) -> Any
+        """Returns the whole body as a json object.
+
+        :return: The JSON deserialized response body
+        :rtype: any
+        :raises json.decoder.JSONDecodeError or ValueError (in python 2.7) if object is not JSON decodable:
+        """
+        # this will trigger errors if response is not read in
+        self.content  # pylint: disable=pointless-statement
+        if not self._json:
+            self._json = loads(self.text())
+        return self._json
+
+    def raise_for_status(self):
+        # type: (...) -> None
+        """Raises an HttpResponseError if the response has an error status code.
+
+        If response is good, does nothing.
+        """
+        if cast(int, self.status_code) >= 400:
+            raise HttpResponseError(response=self)
+
+    @property
+    def content(self):
+        # type: (...) -> bytes
+        """Return the response's content in bytes."""
+        if self._content is None:
+            raise ResponseNotReadError(self)
+        return self._content
+
+    def __repr__(self):
+        # type: (...) -> str
+        content_type_str = (
+            ", Content-Type: {}".format(self.content_type) if self.content_type else ""
+        )
+        return "<HttpResponse: {} {}{}>".format(
+            self.status_code, self.reason, content_type_str
+        )
+
+class HttpResponseImpl(_HttpResponseBaseImpl, _HttpResponse):
+    """HttpResponseImpl built on top of our HttpResponse protocol class.
+
+    Helper impl for creating our transport responses
+    """
+
+    def __enter__(self):
+        # type: (...) -> HttpResponseImpl
+        return self
+
+    def close(self):
+        # type: (...) -> None
+        self._is_closed = True
+        self._internal_response.close()
+
+    def __exit__(self, *args):
+        # type: (...) -> None
+        self.close()
+
+    def read(self):
+        # type: (...) -> bytes
+        """
+        Read the response's bytes.
+
+        """
+        if self._content is None:
+            self._content = b"".join(self.iter_bytes())
+        return self.content
+
+    def iter_text(self):
+        # type: () -> Iterator[str]
+        """Iterate over the response text
+        """
+        for byte in self.iter_bytes():
+            text = byte.decode(self.encoding or "utf-8")
+            yield text
+
+    def iter_lines(self):
+        # type: () -> Iterator[str]
+        for text in self.iter_text():
+            lines = parse_lines_from_text(text)
+            for line in lines:
+                yield line
+
+    def _close_stream(self):
+        # type: (...) -> None
+        self._is_stream_consumed = True
+        self.close()