[rest] have pipelines switch over to rest

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.py

@@ -39,7 +39,6 @@
     RequestIdPolicy,
     RetryPolicy,
 )
-from .pipeline._tools import to_rest_response as _to_rest_response
 
 try:
     from typing import TYPE_CHECKING
@@ -65,17 +64,6 @@
 
 _LOGGER = logging.getLogger(__name__)
 
-def _prepare_request(request):
-    # returns the request ready to run through pipelines
-    # and a bool telling whether we ended up converting it
-    rest_request = False
-    try:
-        request_to_run = request._to_pipeline_transport_request()  # pylint: disable=protected-access
-        rest_request = True
-    except AttributeError:
-        request_to_run = request
-    return rest_request, request_to_run
-
 class PipelineClient(PipelineClientBase):
     """Service client core methods.
 
@@ -203,22 +191,9 @@ def send_request(self, request, **kwargs):
         :keyword bool stream: Whether the response payload will be streamed. Defaults to False.
         :return: The response of your network call. Does not do error handling on your response.
         :rtype: ~azure.core.rest.HttpResponse
-        # """
-        rest_request, request_to_run = _prepare_request(request)
+        """
         return_pipeline_response = kwargs.pop("_return_pipeline_response", False)
-        pipeline_response = self._pipeline.run(request_to_run, **kwargs)  # pylint: disable=protected-access
-        response = pipeline_response.http_response
-        if rest_request:
-            response = _to_rest_response(response)
-            try:
-                if not kwargs.get("stream", False):
-                    response.read()
-                    response.close()
-            except Exception as exc:
-                response.close()
-                raise exc
+        pipeline_response = self._pipeline.run(request, **kwargs)  # pylint: disable=protected-access
         if return_pipeline_response:
-            pipeline_response.http_response = response
-            pipeline_response.http_request = request
             return pipeline_response
-        return response
+        return pipeline_response.http_response

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
@@ -37,8 +37,6 @@
     RequestIdPolicy,
     AsyncRetryPolicy,
 )
-from ._pipeline_client import _prepare_request
-from .pipeline._tools_async import to_rest_response as _to_rest_response
 
 try:
     from typing import TYPE_CHECKING, TypeVar
@@ -63,6 +61,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 +144,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 +153,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 +162,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]
@@ -175,30 +193,13 @@ def _build_pipeline(self, config, **kwargs): # pylint: disable=no-self-use
         return AsyncPipeline(transport, policies)
 
     async def _make_pipeline_call(self, request, **kwargs):
-        rest_request, request_to_run = _prepare_request(request)
         return_pipeline_response = kwargs.pop("_return_pipeline_response", False)
         pipeline_response = await self._pipeline.run(
-            request_to_run, **kwargs  # pylint: disable=protected-access
+            request, **kwargs  # pylint: disable=protected-access
         )
-        response = pipeline_response.http_response
-        if rest_request:
-            rest_response = _to_rest_response(response)
-            if not kwargs.get("stream"):
-                try:
-                    # in this case, the pipeline transport response already called .load_body(), so
-                    # 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.close()
-                except Exception as exc:
-                    await rest_response.close()
-                    raise exc
-            response = rest_response
         if return_pipeline_response:
-            pipeline_response.http_response = response
-            pipeline_response.http_request = request
             return pipeline_response
-        return response
+        return pipeline_response.http_response
 
     def send_request(
         self,
@@ -223,6 +224,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

@@ -33,39 +33,11 @@ def await_result(func, *args, **kwargs):
         )
     return result
 
-def to_rest_request(pipeline_transport_request):
-    from ..rest import HttpRequest as RestHttpRequest
-    return RestHttpRequest(
-        method=pipeline_transport_request.method,
-        url=pipeline_transport_request.url,
-        headers=pipeline_transport_request.headers,
-        files=pipeline_transport_request.files,
-        data=pipeline_transport_request.data
-    )
-
-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
-    response = response_type(
-        request=to_rest_request(pipeline_transport_response.request),
-        internal_response=pipeline_transport_response.internal_response,
-    )
-    response._connection_data_block_size = pipeline_transport_response.block_size  # pylint: disable=protected-access
-    return response
-
-def get_block_size(response):
-    try:
-        return response._connection_data_block_size  # pylint: disable=protected-access
-    except AttributeError:
-        return response.block_size
-
-def get_internal_response(response):
+def read_in_response(response, is_stream_response):
     try:
-        return response._internal_response  # pylint: disable=protected-access
-    except AttributeError:
-        return response.internal_response
+        if not is_stream_response:
+            response.read()
+            response.close()
+    except Exception as exc:
+        response.close()
+        raise exc

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

@@ -23,7 +23,7 @@
 # IN THE SOFTWARE.
 #
 # --------------------------------------------------------------------------
-from ._tools import to_rest_request
+from typing import Optional
 
 async def await_result(func, *args, **kwargs):
     """If func returns an awaitable, await it."""
@@ -33,36 +33,11 @@ async def await_result(func, *args, **kwargs):
         return await result  # type: ignore
     return result
 
-def _get_response_type(pipeline_transport_response):
+async def read_in_response(response, is_stream_response: Optional[bool]) -> None:
     try:
-        from .transport import AioHttpTransportResponse
-        from ..rest._aiohttp import RestAioHttpTransportResponse
-        if isinstance(pipeline_transport_response, AioHttpTransportResponse):
-            return RestAioHttpTransportResponse
-    except ImportError:
-        pass
-    try:
-        from .transport import AsyncioRequestsTransportResponse
-        from ..rest._requests_asyncio import RestAsyncioRequestsTransportResponse
-        if isinstance(pipeline_transport_response, AsyncioRequestsTransportResponse):
-            return RestAsyncioRequestsTransportResponse
-    except ImportError:
-        pass
-    try:
-        from .transport import TrioRequestsTransportResponse
-        from ..rest._requests_trio import RestTrioRequestsTransportResponse
-        if isinstance(pipeline_transport_response, TrioRequestsTransportResponse):
-            return RestTrioRequestsTransportResponse
-    except ImportError:
-        pass
-    from ..rest import AsyncHttpResponse
-    return AsyncHttpResponse
-
-def to_rest_response(pipeline_transport_response):
-    response_type = _get_response_type(pipeline_transport_response)
-    response = response_type(
-        request=to_rest_request(pipeline_transport_response.request),
-        internal_response=pipeline_transport_response.internal_response,
-    )
-    response._connection_data_block_size = pipeline_transport_response.block_size  # pylint: disable=protected-access
-    return response
+        if not is_stream_response:
+            await response.read()
+            await response.close()
+    except Exception as exc:
+        await response.close()
+        raise exc

sdk/core/azure-core/azure/core/pipeline/policies/_universal.py

@@ -576,6 +576,8 @@ def deserialize_from_http_generics(
             mime_type = "application/json"
 
         # Rely on transport implementation to give me "text()" decoded correctly
+        if hasattr(response, "read"):
+            response.read()
         return cls.deserialize_from_text(response.text(encoding), mime_type, response=response)
 
     def on_request(self, request):