Async (#71)

* Make AzureOperationPoller almost a asyncio.coroutine

* Split polling and initial sent into two parts

* AzureOperationPoller is now a Future

* Switch to async/await + test on real SDK

* Implement async with AsyncPoller

* Remove async hack

* Update to latest general msrestazure update

* Pipeline changes for msrestazure

* Update dep

* Fix incorrect last master merge

* Adapt tests to msrest 0.6.2

* Add async extra in setup

* Ignore async tests on Python < 3.5

* Ignore cov errors from async

* Remove async extra

* msrestazure 0.6.0

Author: lmazuel

Pipfile

@@ -11,4 +11,5 @@ mock = {markers="python_version<='2.7'"}
 httpretty = '*'
 pytest = '*'
 pytest-cov = '*'
+pytest-asyncio = {version = "*", markers="python_version >= '3.5'"}
 pylint = '*'

README.rst

@@ -20,6 +20,19 @@ To install:
 Release History
 ---------------
 
+2018-12-17 Version 0.6.0
+++++++++++++++++++++++++
+
+**Features**
+
+- Implementation of LRO async, based on msrest 0.6.x series (*experimental*)
+
+**Disclaimer**
+
+- This version contains no direct breaking changes, but is bumped to 0.6.x since it requires a breaking change version of msrest.
+
+Thanks to @gison93 for his documentation contribution
+
 2018-11-01 Version 0.5.1
 ++++++++++++++++++++++++
 

dev_requirements.txt

@@ -3,3 +3,4 @@ mock;python_version<="2.7"
 httpretty
 pytest
 pytest-cov
+pytest-asyncio;python_full_version>="3.5.2"

msrestazure/azure_exceptions.py

@@ -203,15 +203,17 @@ def _get_state(self, content):
         return "Resource state {}".format(state) if state else "none"
 
     def _build_error_message(self, response):
+        # Assume ClientResponse has "body", and otherwise it's a requests.Response
+        content = response.text() if hasattr(response, "body") else response.text
         try:
-            data = response.json()
+            data = json.loads(content)
         except ValueError:
             message = "none"
         else:
             try:
                 message = data.get("message", self._get_state(data))
             except AttributeError: # data is not a dict, but is a requests.Response parsable as JSON
-                message = str(response.text)
+                message = str(content)
         try:
             response.raise_for_status()
         except RequestException as err:

msrestazure/azure_operation.py

@@ -335,9 +335,6 @@ class AzureOperationPoller(object):
     """Initiates long running operation and polls status in separate
     thread.
 
-    This class is used in old SDK and has been replaced. See "polling"
-    submodule now.
-
     :param callable send_cmd: The API request to initiate the operation.
     :param callable update_cmd: The API reuqest to check the status of
         the operation.
@@ -502,7 +499,7 @@ def wait(self, timeout=None):
 
         :param int timeout: Perion of time to wait for the long running
          operation to complete.
-        :raises ~msrestazure.azure_exceptions.CloudError: Server problem with the query.
+        :raises CloudError: Server problem with the query.
         """
         if self._thread is None:
             return

msrestazure/polling/arm_polling.py

@@ -23,14 +23,14 @@
 # IN THE SOFTWARE.
 #
 # --------------------------------------------------------------------------
-import re
+import json
 import time
 try:
     from urlparse import urlparse
 except ImportError:
     from urllib.parse import urlparse
 
-from msrest.exceptions import DeserializationError, ClientException
+from msrest.exceptions import DeserializationError
 from msrest.polling import PollingMethod
 
 from ..azure_exceptions import CloudError
@@ -143,14 +143,29 @@ def _is_empty(self, response):
         """Check if response body contains meaningful content.
 
         :rtype: bool
-        :raises: DeserializationError if response body contains invalid
-         json data.
+        :raises: DeserializationError if response body contains invalid json data.
         """
-        if not response.content:
+        # Assume ClientResponse has "body", and otherwise it's a requests.Response
+        content = response.text() if hasattr(response, "body") else response.text
+        if not content:
             return True
         try:
-            body = response.json()
-            return not body
+            return not json.loads(content)
+        except ValueError:
+            raise DeserializationError(
+                "Error occurred in deserializing the response body.")
+
+    def _as_json(self, response):
+        """Assuming this is not empty, return the content as JSON.
+
+        Result/exceptions is not determined if you call this method without testing _is_empty.
+
+        :raises: DeserializationError if response body contains invalid json data.
+        """
+        # Assume ClientResponse has "body", and otherwise it's a requests.Response
+        content = response.text() if hasattr(response, "body") else response.text
+        try:
+            return json.loads(content)
         except ValueError:
             raise DeserializationError(
                 "Error occurred in deserializing the response body.")
@@ -171,7 +186,7 @@ def _get_async_status(self, response):
         """
         if self._is_empty(response):
             return None
-        body = response.json()
+        body = self._as_json(response)
         return body.get('status')
 
     def _get_provisioning_state(self, response):
@@ -182,7 +197,7 @@ def _get_provisioning_state(self, response):
         """
         if self._is_empty(response):
             return None
-        body = response.json()
+        body = self._as_json(response)
         return body.get("properties", {}).get("provisioningState")
 
     def should_do_final_get(self):

msrestazure/polling/async_arm_polling.py

@@ -0,0 +1,125 @@
+# --------------------------------------------------------------------------
+#
+# 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.
+#
+# --------------------------------------------------------------------------
+import asyncio
+
+from ..azure_exceptions import CloudError
+from .arm_polling import (
+    failed,
+    BadStatus,
+    BadResponse,
+    OperationFailed,
+    ARMPolling
+)
+
+__all__ = ["AsyncARMPolling"]
+
+class AsyncARMPolling(ARMPolling):
+    """A subclass or ARMPolling that redefine "run" as async.
+    """
+
+    async def run(self):
+        try:
+            await self._poll()
+        except BadStatus:
+            self._operation.status = 'Failed'
+            raise CloudError(self._response)
+
+        except BadResponse as err:
+            self._operation.status = 'Failed'
+            raise CloudError(self._response, str(err))
+
+        except OperationFailed:
+            raise CloudError(self._response)
+
+    async def _poll(self):
+        """Poll status of operation so long as operation is incomplete and
+        we have an endpoint to query.
+
+        :param callable update_cmd: The function to call to retrieve the
+         latest status of the long running operation.
+        :raises: OperationFailed if operation status 'Failed' or 'Cancelled'.
+        :raises: BadStatus if response status invalid.
+        :raises: BadResponse if response invalid.
+        """
+
+        while not self.finished():
+            await self._delay()
+            await self.update_status()
+
+        if failed(self._operation.status):
+            raise OperationFailed("Operation failed or cancelled")
+
+        elif self._operation.should_do_final_get():
+            if self._operation.method == 'POST' and self._operation.location_url:
+                final_get_url = self._operation.location_url
+            else:
+                final_get_url = self._operation.initial_response.request.url
+            self._response = await self.request_status(final_get_url)
+            self._operation.get_status_from_resource(self._response)
+
+    async def _delay(self):
+        """Check for a 'retry-after' header to set timeout,
+        otherwise use configured timeout.
+        """
+        if self._response is None:
+            await asyncio.sleep(0)
+        if self._response.headers.get('retry-after'):
+            await asyncio.sleep(int(self._response.headers['retry-after']))
+        else:
+            await asyncio.sleep(self._timeout)
+
+    async def update_status(self):
+        """Update the current status of the LRO.
+        """
+        if self._operation.async_url:
+            self._response = await self.request_status(self._operation.async_url)
+            self._operation.set_async_url_if_present(self._response)
+            self._operation.get_status_from_async(self._response)
+        elif self._operation.location_url:
+            self._response = await self.request_status(self._operation.location_url)
+            self._operation.set_async_url_if_present(self._response)
+            self._operation.get_status_from_location(self._response)
+        elif self._operation.method == "PUT":
+            initial_url = self._operation.initial_response.request.url
+            self._response = await self.request_status(initial_url)
+            self._operation.set_async_url_if_present(self._response)
+            self._operation.get_status_from_resource(self._response)
+        else:
+            raise BadResponse("Unable to find status link for polling.")
+
+    async def request_status(self, status_link):
+        """Do a simple GET to this status link.
+
+        This method re-inject 'x-ms-client-request-id'.
+
+        :rtype: requests.Response
+        """
+        # ARM requires to re-inject 'x-ms-client-request-id' while polling
+        header_parameters = {
+            'x-ms-client-request-id': self._operation.initial_response.request.headers['x-ms-client-request-id']
+        }
+        request = self._client.get(status_link, headers=header_parameters)
+        return await self._client.async_send(request, stream=False, **self._operation_config)

msrestazure/version.py

@@ -25,4 +25,4 @@
 # --------------------------------------------------------------------------
 
 #: version of the package. Use msrestazure.__version__ instead.
-msrestazure_version = "0.5.1"
+msrestazure_version = "0.6.0"

setup.py

@@ -28,7 +28,7 @@
 
 setup(
     name='msrestazure',
-    version='0.5.1',
+    version='0.6.0',
     author='Microsoft Corporation',
     author_email='azpysdkhelp@microsoft.com',
     packages=find_packages(exclude=["tests", "tests.*"]),
@@ -50,7 +50,7 @@
         'License :: OSI Approved :: MIT License',
         'Topic :: Software Development'],
     install_requires=[
-        "msrest>=0.4.28,<2.0.0",
+        "msrest>=0.6.0,<2.0.0",
         "adal>=0.6.0,<2.0.0",
     ],
 )