File: arm_polling.py

package info (click to toggle)
python-msrestazure 0.6.4-4
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 448 kB
  • sloc: python: 3,942; makefile: 8
file content (470 lines) | stat: -rw-r--r-- 17,467 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
# --------------------------------------------------------------------------
#
# 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 json
import time
try:
    from urlparse import urlparse
except ImportError:
    from urllib.parse import urlparse

from msrest.exceptions import DeserializationError
from msrest.polling import PollingMethod

from ..azure_exceptions import CloudError


FINISHED = frozenset(['succeeded', 'canceled', 'failed'])
FAILED = frozenset(['canceled', 'failed'])
SUCCEEDED = frozenset(['succeeded'])

_AZURE_ASYNC_OPERATION_FINAL_STATE = "azure-async-operation"
_LOCATION_FINAL_STATE = "location"

def finished(status):
    if hasattr(status, 'value'):
        status = status.value
    return str(status).lower() in FINISHED


def failed(status):
    if hasattr(status, 'value'):
        status = status.value
    return str(status).lower() in FAILED


def succeeded(status):
    if hasattr(status, 'value'):
        status = status.value
    return str(status).lower() in SUCCEEDED


class BadStatus(Exception):
    pass


class BadResponse(Exception):
    pass


class OperationFailed(Exception):
    pass

def _validate(url):
    """Validate a url.

    :param str url: Polling URL extracted from response header.
    :raises: ValueError if URL has no scheme or host.
    """
    if url is None:
        return
    parsed = urlparse(url)
    if not parsed.scheme or not parsed.netloc:
        raise ValueError("Invalid URL header")

def get_header_url(response, header_name):
    """Get a URL from a header requests.

    :param requests.Response response: REST call response.
    :param str header_name: Header name.
    :returns: URL if not None AND valid, None otherwise
    """
    url = response.headers.get(header_name)
    try:
        _validate(url)
    except ValueError:
        return None
    else:
        return url


class LongRunningOperation(object):
    """LongRunningOperation
    Provides default logic for interpreting operation responses
    and status updates.

    :param requests.Response response: The initial response.
    :param callable deserialization_callback: The deserialization callaback.
    :param dict lro_options: LRO options.
    :param kwargs: Unused for now
    """

    def __init__(self, response, deserialization_callback, lro_options=None, **kwargs):
        self.method = response.request.method
        self.initial_response = response
        self.status = ""
        self.resource = None
        self.deserialization_callback = deserialization_callback
        self.async_url = None
        self.location_url = None
        if lro_options is None:
            lro_options = {
                'final-state-via': _AZURE_ASYNC_OPERATION_FINAL_STATE
            }
        self.lro_options = lro_options

    def _raise_if_bad_http_status_and_method(self, response):
        """Check response status code is valid for a Put or Patch
        request. Must be 200, 201, 202, or 204.

        :raises: BadStatus if invalid status.
        """
        code = response.status_code
        if code in {200, 202} or \
           (code == 201 and self.method in {'PUT', 'PATCH'}) or \
           (code == 204 and self.method in {'DELETE', 'POST'}):
            return
        raise BadStatus(
            "Invalid return status for {!r} operation".format(self.method))

    def _is_empty(self, response):
        """Check if response body contains meaningful content.

        :rtype: bool
        :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
        if not content:
            return True
        try:
            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.")

    def _deserialize(self, response):
        """Attempt to deserialize resource from response.

        :param requests.Response response: latest REST call response.
        """
        return self.deserialization_callback(response)

    def _get_async_status(self, response):
        """Attempt to find status info in response body.

        :param requests.Response response: latest REST call response.
        :rtype: str
        :returns: Status if found, else 'None'.
        """
        if self._is_empty(response):
            return None
        body = self._as_json(response)
        return body.get('status')

    def _get_provisioning_state(self, response):
        """
        Attempt to get provisioning state from resource.
        :param requests.Response response: latest REST call response.
        :returns: Status if found, else 'None'.
        """
        if self._is_empty(response):
            return None
        body = self._as_json(response)
        return body.get("properties", {}).get("provisioningState")

    def should_do_final_get(self):
        """Check whether the polling should end doing a final GET.

        :param requests.Response response: latest REST call response.
        :rtype: bool
        """
        return ((self.async_url or not self.resource) and self.method in {'PUT', 'PATCH'}) \
                or (self.lro_options['final-state-via'] == _LOCATION_FINAL_STATE and self.location_url and self.async_url and self.method == 'POST')

    def set_initial_status(self, response):
        """Process first response after initiating long running
        operation and set self.status attribute.

        :param requests.Response response: initial REST call response.
        """
        self._raise_if_bad_http_status_and_method(response)

        if self._is_empty(response):
            self.resource = None
        else:
            try:
                self.resource = self._deserialize(response)
            except DeserializationError:
                self.resource = None

        self.set_async_url_if_present(response)

        if response.status_code in {200, 201, 202, 204}:
            if self.async_url or self.location_url or response.status_code == 202:
                self.status = 'InProgress'
            elif response.status_code == 201:
                status = self._get_provisioning_state(response)
                self.status = status or 'InProgress'
            elif response.status_code == 200:
                status = self._get_provisioning_state(response)
                self.status = status or 'Succeeded'
            elif response.status_code == 204:
                self.status = 'Succeeded'
                self.resource = None
            else:
                raise OperationFailed("Invalid status found")
            return
        raise OperationFailed("Operation failed or cancelled")

    def get_status_from_location(self, response):
        """Process the latest status update retrieved from a 'location'
        header.

        :param requests.Response response: latest REST call response.
        :raises: BadResponse if response has no body and not status 202.
        """
        self._raise_if_bad_http_status_and_method(response)
        code = response.status_code
        if code == 202:
            self.status = "InProgress"
        else:
            self.status = 'Succeeded'
            if self._is_empty(response):
                self.resource = None
            else:
                self.resource = self._deserialize(response)

    def get_status_from_resource(self, response):
        """Process the latest status update retrieved from the same URL as
        the previous request.

        :param requests.Response response: latest REST call response.
        :raises: BadResponse if status not 200 or 204.
        """
        self._raise_if_bad_http_status_and_method(response)
        if self._is_empty(response):
            raise BadResponse('The response from long running operation '
                              'does not contain a body.')

        status = self._get_provisioning_state(response)
        self.status = status or 'Succeeded'

        self.parse_resource(response)

    def parse_resource(self, response):
        """Assuming this response is a resource, use the deserialization callback to parse it.
        If body is empty, assuming no resource to return.
        """
        self._raise_if_bad_http_status_and_method(response)
        if not self._is_empty(response):
            self.resource = self._deserialize(response)
        else:
            self.resource = None

    def get_status_from_async(self, response):
        """Process the latest status update retrieved from a
        'azure-asyncoperation' header.

        :param requests.Response response: latest REST call response.
        :raises: BadResponse if response has no body, or body does not
         contain status.
        """
        self._raise_if_bad_http_status_and_method(response)
        if self._is_empty(response):
            raise BadResponse('The response from long running operation '
                              'does not contain a body.')

        self.status = self._get_async_status(response)
        if not self.status:
            raise BadResponse("No status found in body")

        # Status can contains information, see ARM spec:
        # https://github.com/Azure/azure-resource-manager-rpc/blob/master/v1.0/Addendum.md#operation-resource-format
        # "properties": {
        # /\* The resource provider can choose the values here, but it should only be
        #   returned on a successful operation (status being "Succeeded"). \*/
        #},
        # So try to parse it
        try:
            self.resource = self._deserialize(response)
        except Exception:
            self.resource = None

    def set_async_url_if_present(self, response):
        async_url = get_header_url(response, 'azure-asyncoperation')
        if async_url:
            self.async_url = async_url
        location_url = get_header_url(response, 'location')
        if location_url:
            self.location_url = location_url

    def get_status_link(self):
        if self.async_url:
            return self.async_url
        elif self.location_url:
            return self.location_url
        elif self.method == "PUT":
            return self.initial_response.request.url
        else:
            raise BadResponse("Unable to find a valid status link for polling")


class ARMPolling(PollingMethod):

    def __init__(self, timeout=30, lro_options=None, **operation_config):
        self._timeout = timeout
        self._operation = None # Will hold an instance of LongRunningOperation
        self._response = None  # Will hold latest received response
        self._operation_config = operation_config
        self._lro_options = lro_options

    def status(self):
        """Return the current status as a string.
        :rtype: str
        """
        if not self._operation:
            raise ValueError("set_initial_status was never called. Did you give this instance to a poller?")
        return self._operation.status

    def finished(self):
        """Is this polling finished?
        :rtype: bool
        """
        return finished(self.status())

    def resource(self):
        """Return the built resource.
        """
        return self._operation.resource

    def initialize(self, client, initial_response, deserialization_callback):
        """Set the initial status of this LRO.

        :param initial_response: The initial response of the poller
        :raises: CloudError if initial status is incorrect LRO state
        """
        self._client = client
        self._response = initial_response
        self._operation = LongRunningOperation(initial_response, deserialization_callback, self._lro_options)
        try:
            self._operation.set_initial_status(initial_response)
        except BadStatus:
            self._operation.status = 'Failed'
            raise CloudError(initial_response)
        except BadResponse as err:
            self._operation.status = 'Failed'
            raise CloudError(initial_response, str(err))
        except OperationFailed:
            raise CloudError(initial_response)

    def run(self):
        try:
            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)

    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():
            self._delay()
            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 = self.request_status(final_get_url)
            self._operation.parse_resource(self._response)

    def _delay(self):
        """Check for a 'retry-after' header to set timeout,
        otherwise use configured timeout.
        """
        if self._response is None:
            return
        if self._response.headers.get('retry-after'):
            time.sleep(int(self._response.headers['retry-after']))
        else:
            time.sleep(self._timeout)

    def update_status(self):
        """Update the current status of the LRO.
        """
        if self._operation.async_url:
            self._response = 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 = 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 = 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.")

    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
        """
        request = self._client.get(status_link)
        # 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']
        }
        return self._client.send(request, header_parameters, stream=False, **self._operation_config)