1# coding=utf-8
2# --------------------------------------------------------------------------
3# Copyright (c) Microsoft Corporation. All rights reserved.
4# Licensed under the MIT License. See License.txt in the project root for license information.
5# Code generated by Microsoft (R) AutoRest Code Generator.
6# Changes may cause incorrect behavior and will be lost if the code is regenerated.
7# --------------------------------------------------------------------------
8from typing import Any, Callable, Dict, Generic, Optional, TypeVar, Union
9import warnings
10
11from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error
12from azure.core.pipeline import PipelineResponse
13from azure.core.pipeline.transport import AsyncHttpResponse, HttpRequest
14from azure.core.polling import AsyncLROPoller, AsyncNoPolling, AsyncPollingMethod
15from azure.mgmt.core.exceptions import ARMErrorFormat
16from azure.mgmt.core.polling.async_arm_polling import AsyncARMPolling
17
18from ... import models as _models
19
20T = TypeVar('T')
21ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]]
22
23class VirtualMachineExtensionsOperations:
24    """VirtualMachineExtensionsOperations async operations.
25
26    You should not instantiate this class directly. Instead, you should create a Client instance that
27    instantiates it for you and attaches it as an attribute.
28
29    :ivar models: Alias to model classes used in this operation group.
30    :type models: ~azure.mgmt.compute.v2019_03_01.models
31    :param client: Client for service requests.
32    :param config: Configuration of service client.
33    :param serializer: An object model serializer.
34    :param deserializer: An object model deserializer.
35    """
36
37    models = _models
38
39    def __init__(self, client, config, serializer, deserializer) -> None:
40        self._client = client
41        self._serialize = serializer
42        self._deserialize = deserializer
43        self._config = config
44
45    async def _create_or_update_initial(
46        self,
47        resource_group_name: str,
48        vm_name: str,
49        vm_extension_name: str,
50        extension_parameters: "_models.VirtualMachineExtension",
51        **kwargs: Any
52    ) -> "_models.VirtualMachineExtension":
53        cls = kwargs.pop('cls', None)  # type: ClsType["_models.VirtualMachineExtension"]
54        error_map = {
55            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
56        }
57        error_map.update(kwargs.pop('error_map', {}))
58        api_version = "2019-03-01"
59        content_type = kwargs.pop("content_type", "application/json")
60        accept = "application/json"
61
62        # Construct URL
63        url = self._create_or_update_initial.metadata['url']  # type: ignore
64        path_format_arguments = {
65            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
66            'vmName': self._serialize.url("vm_name", vm_name, 'str'),
67            'vmExtensionName': self._serialize.url("vm_extension_name", vm_extension_name, 'str'),
68            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
69        }
70        url = self._client.format_url(url, **path_format_arguments)
71
72        # Construct parameters
73        query_parameters = {}  # type: Dict[str, Any]
74        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
75
76        # Construct headers
77        header_parameters = {}  # type: Dict[str, Any]
78        header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str')
79        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
80
81        body_content_kwargs = {}  # type: Dict[str, Any]
82        body_content = self._serialize.body(extension_parameters, 'VirtualMachineExtension')
83        body_content_kwargs['content'] = body_content
84        request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs)
85        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
86        response = pipeline_response.http_response
87
88        if response.status_code not in [200, 201]:
89            map_error(status_code=response.status_code, response=response, error_map=error_map)
90            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
91
92        if response.status_code == 200:
93            deserialized = self._deserialize('VirtualMachineExtension', pipeline_response)
94
95        if response.status_code == 201:
96            deserialized = self._deserialize('VirtualMachineExtension', pipeline_response)
97
98        if cls:
99            return cls(pipeline_response, deserialized, {})
100
101        return deserialized
102    _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}/extensions/{vmExtensionName}'}  # type: ignore
103
104    async def begin_create_or_update(
105        self,
106        resource_group_name: str,
107        vm_name: str,
108        vm_extension_name: str,
109        extension_parameters: "_models.VirtualMachineExtension",
110        **kwargs: Any
111    ) -> AsyncLROPoller["_models.VirtualMachineExtension"]:
112        """The operation to create or update the extension.
113
114        :param resource_group_name: The name of the resource group.
115        :type resource_group_name: str
116        :param vm_name: The name of the virtual machine where the extension should be created or
117         updated.
118        :type vm_name: str
119        :param vm_extension_name: The name of the virtual machine extension.
120        :type vm_extension_name: str
121        :param extension_parameters: Parameters supplied to the Create Virtual Machine Extension
122         operation.
123        :type extension_parameters: ~azure.mgmt.compute.v2019_03_01.models.VirtualMachineExtension
124        :keyword callable cls: A custom type or function that will be passed the direct response
125        :keyword str continuation_token: A continuation token to restart a poller from a saved state.
126        :keyword polling: By default, your polling method will be AsyncARMPolling.
127         Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy.
128        :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod
129        :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present.
130        :return: An instance of AsyncLROPoller that returns either VirtualMachineExtension or the result of cls(response)
131        :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.compute.v2019_03_01.models.VirtualMachineExtension]
132        :raises ~azure.core.exceptions.HttpResponseError:
133        """
134        polling = kwargs.pop('polling', True)  # type: Union[bool, AsyncPollingMethod]
135        cls = kwargs.pop('cls', None)  # type: ClsType["_models.VirtualMachineExtension"]
136        lro_delay = kwargs.pop(
137            'polling_interval',
138            self._config.polling_interval
139        )
140        cont_token = kwargs.pop('continuation_token', None)  # type: Optional[str]
141        if cont_token is None:
142            raw_result = await self._create_or_update_initial(
143                resource_group_name=resource_group_name,
144                vm_name=vm_name,
145                vm_extension_name=vm_extension_name,
146                extension_parameters=extension_parameters,
147                cls=lambda x,y,z: x,
148                **kwargs
149            )
150
151        kwargs.pop('error_map', None)
152        kwargs.pop('content_type', None)
153
154        def get_long_running_output(pipeline_response):
155            deserialized = self._deserialize('VirtualMachineExtension', pipeline_response)
156
157            if cls:
158                return cls(pipeline_response, deserialized, {})
159            return deserialized
160
161        path_format_arguments = {
162            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
163            'vmName': self._serialize.url("vm_name", vm_name, 'str'),
164            'vmExtensionName': self._serialize.url("vm_extension_name", vm_extension_name, 'str'),
165            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
166        }
167
168        if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments,  **kwargs)
169        elif polling is False: polling_method = AsyncNoPolling()
170        else: polling_method = polling
171        if cont_token:
172            return AsyncLROPoller.from_continuation_token(
173                polling_method=polling_method,
174                continuation_token=cont_token,
175                client=self._client,
176                deserialization_callback=get_long_running_output
177            )
178        else:
179            return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method)
180    begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}/extensions/{vmExtensionName}'}  # type: ignore
181
182    async def _update_initial(
183        self,
184        resource_group_name: str,
185        vm_name: str,
186        vm_extension_name: str,
187        extension_parameters: "_models.VirtualMachineExtensionUpdate",
188        **kwargs: Any
189    ) -> "_models.VirtualMachineExtension":
190        cls = kwargs.pop('cls', None)  # type: ClsType["_models.VirtualMachineExtension"]
191        error_map = {
192            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
193        }
194        error_map.update(kwargs.pop('error_map', {}))
195        api_version = "2019-03-01"
196        content_type = kwargs.pop("content_type", "application/json")
197        accept = "application/json"
198
199        # Construct URL
200        url = self._update_initial.metadata['url']  # type: ignore
201        path_format_arguments = {
202            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
203            'vmName': self._serialize.url("vm_name", vm_name, 'str'),
204            'vmExtensionName': self._serialize.url("vm_extension_name", vm_extension_name, 'str'),
205            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
206        }
207        url = self._client.format_url(url, **path_format_arguments)
208
209        # Construct parameters
210        query_parameters = {}  # type: Dict[str, Any]
211        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
212
213        # Construct headers
214        header_parameters = {}  # type: Dict[str, Any]
215        header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str')
216        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
217
218        body_content_kwargs = {}  # type: Dict[str, Any]
219        body_content = self._serialize.body(extension_parameters, 'VirtualMachineExtensionUpdate')
220        body_content_kwargs['content'] = body_content
221        request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs)
222        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
223        response = pipeline_response.http_response
224
225        if response.status_code not in [200]:
226            map_error(status_code=response.status_code, response=response, error_map=error_map)
227            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
228
229        deserialized = self._deserialize('VirtualMachineExtension', pipeline_response)
230
231        if cls:
232            return cls(pipeline_response, deserialized, {})
233
234        return deserialized
235    _update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}/extensions/{vmExtensionName}'}  # type: ignore
236
237    async def begin_update(
238        self,
239        resource_group_name: str,
240        vm_name: str,
241        vm_extension_name: str,
242        extension_parameters: "_models.VirtualMachineExtensionUpdate",
243        **kwargs: Any
244    ) -> AsyncLROPoller["_models.VirtualMachineExtension"]:
245        """The operation to update the extension.
246
247        :param resource_group_name: The name of the resource group.
248        :type resource_group_name: str
249        :param vm_name: The name of the virtual machine where the extension should be updated.
250        :type vm_name: str
251        :param vm_extension_name: The name of the virtual machine extension.
252        :type vm_extension_name: str
253        :param extension_parameters: Parameters supplied to the Update Virtual Machine Extension
254         operation.
255        :type extension_parameters: ~azure.mgmt.compute.v2019_03_01.models.VirtualMachineExtensionUpdate
256        :keyword callable cls: A custom type or function that will be passed the direct response
257        :keyword str continuation_token: A continuation token to restart a poller from a saved state.
258        :keyword polling: By default, your polling method will be AsyncARMPolling.
259         Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy.
260        :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod
261        :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present.
262        :return: An instance of AsyncLROPoller that returns either VirtualMachineExtension or the result of cls(response)
263        :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.compute.v2019_03_01.models.VirtualMachineExtension]
264        :raises ~azure.core.exceptions.HttpResponseError:
265        """
266        polling = kwargs.pop('polling', True)  # type: Union[bool, AsyncPollingMethod]
267        cls = kwargs.pop('cls', None)  # type: ClsType["_models.VirtualMachineExtension"]
268        lro_delay = kwargs.pop(
269            'polling_interval',
270            self._config.polling_interval
271        )
272        cont_token = kwargs.pop('continuation_token', None)  # type: Optional[str]
273        if cont_token is None:
274            raw_result = await self._update_initial(
275                resource_group_name=resource_group_name,
276                vm_name=vm_name,
277                vm_extension_name=vm_extension_name,
278                extension_parameters=extension_parameters,
279                cls=lambda x,y,z: x,
280                **kwargs
281            )
282
283        kwargs.pop('error_map', None)
284        kwargs.pop('content_type', None)
285
286        def get_long_running_output(pipeline_response):
287            deserialized = self._deserialize('VirtualMachineExtension', pipeline_response)
288
289            if cls:
290                return cls(pipeline_response, deserialized, {})
291            return deserialized
292
293        path_format_arguments = {
294            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
295            'vmName': self._serialize.url("vm_name", vm_name, 'str'),
296            'vmExtensionName': self._serialize.url("vm_extension_name", vm_extension_name, 'str'),
297            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
298        }
299
300        if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments,  **kwargs)
301        elif polling is False: polling_method = AsyncNoPolling()
302        else: polling_method = polling
303        if cont_token:
304            return AsyncLROPoller.from_continuation_token(
305                polling_method=polling_method,
306                continuation_token=cont_token,
307                client=self._client,
308                deserialization_callback=get_long_running_output
309            )
310        else:
311            return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method)
312    begin_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}/extensions/{vmExtensionName}'}  # type: ignore
313
314    async def _delete_initial(
315        self,
316        resource_group_name: str,
317        vm_name: str,
318        vm_extension_name: str,
319        **kwargs: Any
320    ) -> None:
321        cls = kwargs.pop('cls', None)  # type: ClsType[None]
322        error_map = {
323            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
324        }
325        error_map.update(kwargs.pop('error_map', {}))
326        api_version = "2019-03-01"
327
328        # Construct URL
329        url = self._delete_initial.metadata['url']  # type: ignore
330        path_format_arguments = {
331            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
332            'vmName': self._serialize.url("vm_name", vm_name, 'str'),
333            'vmExtensionName': self._serialize.url("vm_extension_name", vm_extension_name, 'str'),
334            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
335        }
336        url = self._client.format_url(url, **path_format_arguments)
337
338        # Construct parameters
339        query_parameters = {}  # type: Dict[str, Any]
340        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
341
342        # Construct headers
343        header_parameters = {}  # type: Dict[str, Any]
344
345        request = self._client.delete(url, query_parameters, header_parameters)
346        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
347        response = pipeline_response.http_response
348
349        if response.status_code not in [200, 202, 204]:
350            map_error(status_code=response.status_code, response=response, error_map=error_map)
351            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
352
353        if cls:
354            return cls(pipeline_response, None, {})
355
356    _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}/extensions/{vmExtensionName}'}  # type: ignore
357
358    async def begin_delete(
359        self,
360        resource_group_name: str,
361        vm_name: str,
362        vm_extension_name: str,
363        **kwargs: Any
364    ) -> AsyncLROPoller[None]:
365        """The operation to delete the extension.
366
367        :param resource_group_name: The name of the resource group.
368        :type resource_group_name: str
369        :param vm_name: The name of the virtual machine where the extension should be deleted.
370        :type vm_name: str
371        :param vm_extension_name: The name of the virtual machine extension.
372        :type vm_extension_name: str
373        :keyword callable cls: A custom type or function that will be passed the direct response
374        :keyword str continuation_token: A continuation token to restart a poller from a saved state.
375        :keyword polling: By default, your polling method will be AsyncARMPolling.
376         Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy.
377        :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod
378        :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present.
379        :return: An instance of AsyncLROPoller that returns either None or the result of cls(response)
380        :rtype: ~azure.core.polling.AsyncLROPoller[None]
381        :raises ~azure.core.exceptions.HttpResponseError:
382        """
383        polling = kwargs.pop('polling', True)  # type: Union[bool, AsyncPollingMethod]
384        cls = kwargs.pop('cls', None)  # type: ClsType[None]
385        lro_delay = kwargs.pop(
386            'polling_interval',
387            self._config.polling_interval
388        )
389        cont_token = kwargs.pop('continuation_token', None)  # type: Optional[str]
390        if cont_token is None:
391            raw_result = await self._delete_initial(
392                resource_group_name=resource_group_name,
393                vm_name=vm_name,
394                vm_extension_name=vm_extension_name,
395                cls=lambda x,y,z: x,
396                **kwargs
397            )
398
399        kwargs.pop('error_map', None)
400        kwargs.pop('content_type', None)
401
402        def get_long_running_output(pipeline_response):
403            if cls:
404                return cls(pipeline_response, None, {})
405
406        path_format_arguments = {
407            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
408            'vmName': self._serialize.url("vm_name", vm_name, 'str'),
409            'vmExtensionName': self._serialize.url("vm_extension_name", vm_extension_name, 'str'),
410            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
411        }
412
413        if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments,  **kwargs)
414        elif polling is False: polling_method = AsyncNoPolling()
415        else: polling_method = polling
416        if cont_token:
417            return AsyncLROPoller.from_continuation_token(
418                polling_method=polling_method,
419                continuation_token=cont_token,
420                client=self._client,
421                deserialization_callback=get_long_running_output
422            )
423        else:
424            return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method)
425    begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}/extensions/{vmExtensionName}'}  # type: ignore
426
427    async def get(
428        self,
429        resource_group_name: str,
430        vm_name: str,
431        vm_extension_name: str,
432        expand: Optional[str] = None,
433        **kwargs: Any
434    ) -> "_models.VirtualMachineExtension":
435        """The operation to get the extension.
436
437        :param resource_group_name: The name of the resource group.
438        :type resource_group_name: str
439        :param vm_name: The name of the virtual machine containing the extension.
440        :type vm_name: str
441        :param vm_extension_name: The name of the virtual machine extension.
442        :type vm_extension_name: str
443        :param expand: The expand expression to apply on the operation.
444        :type expand: str
445        :keyword callable cls: A custom type or function that will be passed the direct response
446        :return: VirtualMachineExtension, or the result of cls(response)
447        :rtype: ~azure.mgmt.compute.v2019_03_01.models.VirtualMachineExtension
448        :raises: ~azure.core.exceptions.HttpResponseError
449        """
450        cls = kwargs.pop('cls', None)  # type: ClsType["_models.VirtualMachineExtension"]
451        error_map = {
452            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
453        }
454        error_map.update(kwargs.pop('error_map', {}))
455        api_version = "2019-03-01"
456        accept = "application/json"
457
458        # Construct URL
459        url = self.get.metadata['url']  # type: ignore
460        path_format_arguments = {
461            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
462            'vmName': self._serialize.url("vm_name", vm_name, 'str'),
463            'vmExtensionName': self._serialize.url("vm_extension_name", vm_extension_name, 'str'),
464            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
465        }
466        url = self._client.format_url(url, **path_format_arguments)
467
468        # Construct parameters
469        query_parameters = {}  # type: Dict[str, Any]
470        if expand is not None:
471            query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
472        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
473
474        # Construct headers
475        header_parameters = {}  # type: Dict[str, Any]
476        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
477
478        request = self._client.get(url, query_parameters, header_parameters)
479        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
480        response = pipeline_response.http_response
481
482        if response.status_code not in [200]:
483            map_error(status_code=response.status_code, response=response, error_map=error_map)
484            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
485
486        deserialized = self._deserialize('VirtualMachineExtension', pipeline_response)
487
488        if cls:
489            return cls(pipeline_response, deserialized, {})
490
491        return deserialized
492    get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}/extensions/{vmExtensionName}'}  # type: ignore
493
494    async def list(
495        self,
496        resource_group_name: str,
497        vm_name: str,
498        expand: Optional[str] = None,
499        **kwargs: Any
500    ) -> "_models.VirtualMachineExtensionsListResult":
501        """The operation to get all extensions of a Virtual Machine.
502
503        :param resource_group_name: The name of the resource group.
504        :type resource_group_name: str
505        :param vm_name: The name of the virtual machine containing the extension.
506        :type vm_name: str
507        :param expand: The expand expression to apply on the operation.
508        :type expand: str
509        :keyword callable cls: A custom type or function that will be passed the direct response
510        :return: VirtualMachineExtensionsListResult, or the result of cls(response)
511        :rtype: ~azure.mgmt.compute.v2019_03_01.models.VirtualMachineExtensionsListResult
512        :raises: ~azure.core.exceptions.HttpResponseError
513        """
514        cls = kwargs.pop('cls', None)  # type: ClsType["_models.VirtualMachineExtensionsListResult"]
515        error_map = {
516            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
517        }
518        error_map.update(kwargs.pop('error_map', {}))
519        api_version = "2019-03-01"
520        accept = "application/json"
521
522        # Construct URL
523        url = self.list.metadata['url']  # type: ignore
524        path_format_arguments = {
525            'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'),
526            'vmName': self._serialize.url("vm_name", vm_name, 'str'),
527            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
528        }
529        url = self._client.format_url(url, **path_format_arguments)
530
531        # Construct parameters
532        query_parameters = {}  # type: Dict[str, Any]
533        if expand is not None:
534            query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
535        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
536
537        # Construct headers
538        header_parameters = {}  # type: Dict[str, Any]
539        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
540
541        request = self._client.get(url, query_parameters, header_parameters)
542        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
543        response = pipeline_response.http_response
544
545        if response.status_code not in [200]:
546            map_error(status_code=response.status_code, response=response, error_map=error_map)
547            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
548
549        deserialized = self._deserialize('VirtualMachineExtensionsListResult', pipeline_response)
550
551        if cls:
552            return cls(pipeline_response, deserialized, {})
553
554        return deserialized
555    list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachines/{vmName}/extensions'}  # type: ignore
556