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