Multipart/mixed in azure-core

sdk/core/azure-core/HISTORY.md

@@ -9,6 +9,14 @@
 
 - Tracing: network span context is available with the TRACING_CONTEXT in pipeline response  #7252
 - Tracing: Span contract now has `kind`, `traceparent` and is a context manager  #7252
+- SansIOHTTPPolicy methods can now be coroutines #7497
+- Add multipart/mixed support #7083:
+
+  - HttpRequest now has a "set_multipart_mixed" method to set the parts of this request
+  - HttpRequest now has a "prepare_multipart_body" method to build final body.
+  - HttpResponse now has a "parts" method to return an iterator of parts
+  - AsyncHttpResponse now has a "parts" methods to return an async iterator of parts
+  - Note that multipart/MIXED is a Python 3.x only feature
 
 ### Bug fixes
 

sdk/core/azure-core/README.md

@@ -241,6 +241,11 @@ class HttpRequest(object):
 
     def set_bytes_body(self, data):
         """Set generic bytes as the body of the request."""
+
+    def set_multipart_mixed(self, *requests, **kwargs):
+        """Set requests for a multipart/mixed body.
+        Optionally apply "policies" in kwargs to each request.
+        """
 ```
 
 The HttpResponse object on the other hand will generally have a transport-specific derivative.
@@ -285,6 +290,12 @@ class HttpResponse(object):
         and asynchronous generator.
         """
 
+    def parts(self):
+        """An iterator of parts if content-type is multipart/mixed.
+        For the AsyncHttpResponse object this function will return
+        and asynchronous iterator.
+        """
+
 ```
 
 ### PipelineRequest and PipelineResponse
@@ -344,6 +355,8 @@ def on_exception(self, request):
     """
 ```
 
+SansIOHTTPPolicy methods can be declared as coroutines, but then they can only be used with a AsyncPipeline.
+
 Current provided sans IO policies include:
 ```python
 from azure.core.pipeline.policies import (

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

@@ -25,19 +25,22 @@
 # --------------------------------------------------------------------------
 
 import abc
-from typing import (TypeVar, Any, Dict, Optional, Generic)
+from typing import TypeVar, Generic
 
 try:
     ABC = abc.ABC
-except AttributeError: # Python 2.7, abc exists, but not ABC
-    ABC = abc.ABCMeta('ABC', (object,), {'__slots__': ()})  # type: ignore
+except AttributeError:  # Python 2.7, abc exists, but not ABC
+    ABC = abc.ABCMeta("ABC", (object,), {"__slots__": ()})  # type: ignore
 
 HTTPResponseType = TypeVar("HTTPResponseType")
 HTTPRequestType = TypeVar("HTTPRequestType")
 
 try:
-    from contextlib import AbstractContextManager  # type: ignore #pylint: disable=unused-import
-except ImportError: # Python <= 3.5
+    from contextlib import (  # pylint: disable=unused-import
+        AbstractContextManager,
+    )  # type: ignore
+except ImportError:  # Python <= 3.5
+
     class AbstractContextManager(object):  # type: ignore
         def __enter__(self):
             """Return `self` upon entering the runtime context."""
@@ -60,19 +63,20 @@ class PipelineContext(dict):
     :param transport: The HTTP transport type.
     :param kwargs: Developer-defined keyword arguments.
     """
-    def __init__(self, transport, **kwargs): #pylint: disable=super-init-not-called
+
+    def __init__(self, transport, **kwargs):  # pylint: disable=super-init-not-called
         self.transport = transport
         self.options = kwargs
-        self._protected = ['transport', 'options']
+        self._protected = ["transport", "options"]
 
     def __setitem__(self, key, item):
         if key in self._protected:
-            raise ValueError('Context value {} cannot be overwritten.'.format(key))
+            raise ValueError("Context value {} cannot be overwritten.".format(key))
         return super(PipelineContext, self).__setitem__(key, item)
 
     def __delitem__(self, key):
         if key in self._protected:
-            raise ValueError('Context value {} cannot be deleted.'.format(key))
+            raise ValueError("Context value {} cannot be deleted.".format(key))
         return super(PipelineContext, self).__delitem__(key)
 
     def clear(self):
@@ -93,7 +97,7 @@ def pop(self, *args):
         """Removes specified key and returns the value.
         """
         if args and args[0] in self._protected:
-            raise ValueError('Context value {} cannot be popped.'.format(args[0]))
+            raise ValueError("Context value {} cannot be popped.".format(args[0]))
         return super(PipelineContext, self).pop(*args)
 
 
@@ -108,6 +112,7 @@ class PipelineRequest(Generic[HTTPRequestType]):
     :param context: Contains the context - data persisted between pipeline requests.
     :type context: ~azure.core.pipeline.PipelineContext
     """
+
     def __init__(self, http_request, context):
         # type: (HTTPRequestType, PipelineContext) -> None
         self.http_request = http_request
@@ -131,24 +136,21 @@ class PipelineResponse(Generic[HTTPRequestType, HTTPResponseType]):
     :param context: Contains the context - data persisted between pipeline requests.
     :type context: ~azure.core.pipeline.PipelineContext
     """
+
     def __init__(self, http_request, http_response, context):
         # type: (HTTPRequestType, HTTPResponseType, PipelineContext) -> None
         self.http_request = http_request
         self.http_response = http_response
         self.context = context
 
 
-from .base import Pipeline #pylint: disable=wrong-import-position
+from .base import Pipeline  # pylint: disable=wrong-import-position
 
-__all__ = [
-    'Pipeline',
-    'PipelineRequest',
-    'PipelineResponse',
-    'PipelineContext'
-]
+__all__ = ["Pipeline", "PipelineRequest", "PipelineResponse", "PipelineContext"]
 
 try:
-    from .base_async import AsyncPipeline #pylint: disable=unused-import
-    __all__.append('AsyncPipeline')
+    from .base_async import AsyncPipeline  # pylint: disable=unused-import
+
+    __all__.append("AsyncPipeline")
 except (SyntaxError, ImportError):
     pass  # Asynchronous pipelines not supported.

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

@@ -25,10 +25,15 @@
 # --------------------------------------------------------------------------
 
 import logging
-from typing import (TYPE_CHECKING, Generic, TypeVar, cast, IO, List, Union, Any, Mapping, Dict, Optional, # pylint: disable=unused-import
-                    Tuple, Callable, Iterator)
-from azure.core.pipeline import AbstractContextManager, PipelineRequest, PipelineResponse, PipelineContext
+from typing import Generic, TypeVar, List, Union, Any
+from azure.core.pipeline import (
+    AbstractContextManager,
+    PipelineRequest,
+    PipelineResponse,
+    PipelineContext,
+)
 from azure.core.pipeline.policies import HTTPPolicy, SansIOHTTPPolicy
+
 HTTPResponseType = TypeVar("HTTPResponseType")
 HTTPRequestType = TypeVar("HTTPRequestType")
 HttpTransportType = TypeVar("HttpTransportType")
@@ -40,8 +45,10 @@
 def _await_result(func, *args, **kwargs):
     """If func returns an awaitable, raise that this runner can't handle it."""
     result = func(*args, **kwargs)
-    if hasattr(result, '__await__'):
-        raise TypeError("Policy {} returned awaitable object in non-async pipeline.".format(func))
+    if hasattr(result, "__await__"):
+        raise TypeError(
+            "Policy {} returned awaitable object in non-async pipeline.".format(func)
+        )
     return result
 
 
@@ -71,7 +78,7 @@ def send(self, request):
         _await_result(self._policy.on_request, request)
         try:
             response = self.next.send(request)
-        except Exception: #pylint: disable=broad-except
+        except Exception:  # pylint: disable=broad-except
             if not _await_result(self._policy.on_exception, request):
                 raise
         else:
@@ -86,6 +93,7 @@ class _TransportRunner(HTTPPolicy):
 
     :param sender: The Http Transport instance.
     """
+
     def __init__(self, sender):
         # type: (HttpTransportType) -> None
         super(_TransportRunner, self).__init__()
@@ -102,7 +110,7 @@ def send(self, request):
         return PipelineResponse(
             request.http_request,
             self._sender.send(request.http_request, **request.context.options),
-            context=request.context
+            context=request.context,
         )
 
 
@@ -123,29 +131,59 @@ class Pipeline(AbstractContextManager, Generic[HTTPRequestType, HTTPResponseType
             :dedent: 4
             :caption: Builds the pipeline for synchronous transport.
     """
+
     def __init__(self, transport, policies=None):
         # type: (HttpTransportType, PoliciesType) -> None
         self._impl_policies = []  # type: List[HTTPPolicy]
         self._transport = transport  # type: ignore
 
-        for policy in (policies or []):
+        for policy in policies or []:
             if isinstance(policy, SansIOHTTPPolicy):
                 self._impl_policies.append(_SansIOHTTPPolicyRunner(policy))
             elif policy:
                 self._impl_policies.append(policy)
-        for index in range(len(self._impl_policies)-1):
-            self._impl_policies[index].next = self._impl_policies[index+1]
+        for index in range(len(self._impl_policies) - 1):
+            self._impl_policies[index].next = self._impl_policies[index + 1]
         if self._impl_policies:
             self._impl_policies[-1].next = _TransportRunner(self._transport)
 
     def __enter__(self):
         # type: () -> Pipeline
-        self._transport.__enter__() # type: ignore
+        self._transport.__enter__()  # type: ignore
         return self
 
     def __exit__(self, *exc_details):  # pylint: disable=arguments-differ
         self._transport.__exit__(*exc_details)
 
+    @staticmethod
+    def _prepare_multipart_mixed_request(request):
+        # type: (HTTPRequestType) -> None
+        """Will execute the multipart policies.
+
+        Does nothing if "set_multipart_mixed" was never called.
+        """
+        multipart_mixed_info = request.multipart_mixed_info  # type: ignore
+        if not multipart_mixed_info:
+            return
+
+        requests = multipart_mixed_info[0]  # type: List[HTTPRequestType]
+        policies = multipart_mixed_info[1]  # type: List[SansIOHTTPPolicy]
+
+        # Apply on_requests concurrently to all requests
+        import concurrent.futures
+
+        def prepare_requests(req):
+            context = PipelineContext(None)
+            pipeline_request = PipelineRequest(req, context)
+            for policy in policies:
+                _await_result(policy.on_request, pipeline_request)
+
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            # List comprehension to raise exceptions if happened
+            [  # pylint: disable=expression-not-assigned
+                _ for _ in executor.map(prepare_requests, requests)
+            ]
+
     def run(self, request, **kwargs):
         # type: (HTTPRequestType, Any) -> PipelineResponse
         """Runs the HTTP Request through the chained policies.
@@ -155,7 +193,15 @@ def run(self, request, **kwargs):
         :return: The PipelineResponse object
         :rtype: ~azure.core.pipeline.PipelineResponse
         """
+        self._prepare_multipart_mixed_request(request)
+        request.prepare_multipart_body()  # type: ignore
         context = PipelineContext(self._transport, **kwargs)
-        pipeline_request = PipelineRequest(request, context) # type: PipelineRequest[HTTPRequestType]
-        first_node = self._impl_policies[0] if self._impl_policies else _TransportRunner(self._transport)
+        pipeline_request = PipelineRequest(
+            request, context
+        )  # type: PipelineRequest[HTTPRequestType]
+        first_node = (
+            self._impl_policies[0]
+            if self._impl_policies
+            else _TransportRunner(self._transport)
+        )
         return first_node.send(pipeline_request)  # type: ignore