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 VirtualMachineScaleSetRollingUpgradesOperations: 24 """VirtualMachineScaleSetRollingUpgradesOperations 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_12_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 _cancel_initial( 46 self, 47 resource_group_name: str, 48 vm_scale_set_name: str, 49 **kwargs: Any 50 ) -> None: 51 cls = kwargs.pop('cls', None) # type: ClsType[None] 52 error_map = { 53 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 54 } 55 error_map.update(kwargs.pop('error_map', {})) 56 api_version = "2019-12-01" 57 58 # Construct URL 59 url = self._cancel_initial.metadata['url'] # type: ignore 60 path_format_arguments = { 61 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 62 'vmScaleSetName': self._serialize.url("vm_scale_set_name", vm_scale_set_name, 'str'), 63 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 64 } 65 url = self._client.format_url(url, **path_format_arguments) 66 67 # Construct parameters 68 query_parameters = {} # type: Dict[str, Any] 69 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 70 71 # Construct headers 72 header_parameters = {} # type: Dict[str, Any] 73 74 request = self._client.post(url, query_parameters, header_parameters) 75 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 76 response = pipeline_response.http_response 77 78 if response.status_code not in [200, 202]: 79 map_error(status_code=response.status_code, response=response, error_map=error_map) 80 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 81 82 if cls: 83 return cls(pipeline_response, None, {}) 84 85 _cancel_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachineScaleSets/{vmScaleSetName}/rollingUpgrades/cancel'} # type: ignore 86 87 async def begin_cancel( 88 self, 89 resource_group_name: str, 90 vm_scale_set_name: str, 91 **kwargs: Any 92 ) -> AsyncLROPoller[None]: 93 """Cancels the current virtual machine scale set rolling upgrade. 94 95 :param resource_group_name: The name of the resource group. 96 :type resource_group_name: str 97 :param vm_scale_set_name: The name of the VM scale set. 98 :type vm_scale_set_name: str 99 :keyword callable cls: A custom type or function that will be passed the direct response 100 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 101 :keyword polling: By default, your polling method will be AsyncARMPolling. 102 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 103 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 104 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 105 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 106 :rtype: ~azure.core.polling.AsyncLROPoller[None] 107 :raises ~azure.core.exceptions.HttpResponseError: 108 """ 109 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 110 cls = kwargs.pop('cls', None) # type: ClsType[None] 111 lro_delay = kwargs.pop( 112 'polling_interval', 113 self._config.polling_interval 114 ) 115 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 116 if cont_token is None: 117 raw_result = await self._cancel_initial( 118 resource_group_name=resource_group_name, 119 vm_scale_set_name=vm_scale_set_name, 120 cls=lambda x,y,z: x, 121 **kwargs 122 ) 123 124 kwargs.pop('error_map', None) 125 kwargs.pop('content_type', None) 126 127 def get_long_running_output(pipeline_response): 128 if cls: 129 return cls(pipeline_response, None, {}) 130 131 path_format_arguments = { 132 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 133 'vmScaleSetName': self._serialize.url("vm_scale_set_name", vm_scale_set_name, 'str'), 134 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 135 } 136 137 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 138 elif polling is False: polling_method = AsyncNoPolling() 139 else: polling_method = polling 140 if cont_token: 141 return AsyncLROPoller.from_continuation_token( 142 polling_method=polling_method, 143 continuation_token=cont_token, 144 client=self._client, 145 deserialization_callback=get_long_running_output 146 ) 147 else: 148 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 149 begin_cancel.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachineScaleSets/{vmScaleSetName}/rollingUpgrades/cancel'} # type: ignore 150 151 async def _start_os_upgrade_initial( 152 self, 153 resource_group_name: str, 154 vm_scale_set_name: str, 155 **kwargs: Any 156 ) -> None: 157 cls = kwargs.pop('cls', None) # type: ClsType[None] 158 error_map = { 159 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 160 } 161 error_map.update(kwargs.pop('error_map', {})) 162 api_version = "2019-12-01" 163 164 # Construct URL 165 url = self._start_os_upgrade_initial.metadata['url'] # type: ignore 166 path_format_arguments = { 167 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 168 'vmScaleSetName': self._serialize.url("vm_scale_set_name", vm_scale_set_name, 'str'), 169 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 170 } 171 url = self._client.format_url(url, **path_format_arguments) 172 173 # Construct parameters 174 query_parameters = {} # type: Dict[str, Any] 175 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 176 177 # Construct headers 178 header_parameters = {} # type: Dict[str, Any] 179 180 request = self._client.post(url, query_parameters, header_parameters) 181 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 182 response = pipeline_response.http_response 183 184 if response.status_code not in [200, 202]: 185 map_error(status_code=response.status_code, response=response, error_map=error_map) 186 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 187 188 if cls: 189 return cls(pipeline_response, None, {}) 190 191 _start_os_upgrade_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachineScaleSets/{vmScaleSetName}/osRollingUpgrade'} # type: ignore 192 193 async def begin_start_os_upgrade( 194 self, 195 resource_group_name: str, 196 vm_scale_set_name: str, 197 **kwargs: Any 198 ) -> AsyncLROPoller[None]: 199 """Starts a rolling upgrade to move all virtual machine scale set instances to the latest 200 available Platform Image OS version. Instances which are already running the latest available 201 OS version are not affected. 202 203 :param resource_group_name: The name of the resource group. 204 :type resource_group_name: str 205 :param vm_scale_set_name: The name of the VM scale set. 206 :type vm_scale_set_name: str 207 :keyword callable cls: A custom type or function that will be passed the direct response 208 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 209 :keyword polling: By default, your polling method will be AsyncARMPolling. 210 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 211 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 212 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 213 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 214 :rtype: ~azure.core.polling.AsyncLROPoller[None] 215 :raises ~azure.core.exceptions.HttpResponseError: 216 """ 217 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 218 cls = kwargs.pop('cls', None) # type: ClsType[None] 219 lro_delay = kwargs.pop( 220 'polling_interval', 221 self._config.polling_interval 222 ) 223 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 224 if cont_token is None: 225 raw_result = await self._start_os_upgrade_initial( 226 resource_group_name=resource_group_name, 227 vm_scale_set_name=vm_scale_set_name, 228 cls=lambda x,y,z: x, 229 **kwargs 230 ) 231 232 kwargs.pop('error_map', None) 233 kwargs.pop('content_type', None) 234 235 def get_long_running_output(pipeline_response): 236 if cls: 237 return cls(pipeline_response, None, {}) 238 239 path_format_arguments = { 240 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 241 'vmScaleSetName': self._serialize.url("vm_scale_set_name", vm_scale_set_name, 'str'), 242 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 243 } 244 245 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 246 elif polling is False: polling_method = AsyncNoPolling() 247 else: polling_method = polling 248 if cont_token: 249 return AsyncLROPoller.from_continuation_token( 250 polling_method=polling_method, 251 continuation_token=cont_token, 252 client=self._client, 253 deserialization_callback=get_long_running_output 254 ) 255 else: 256 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 257 begin_start_os_upgrade.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachineScaleSets/{vmScaleSetName}/osRollingUpgrade'} # type: ignore 258 259 async def _start_extension_upgrade_initial( 260 self, 261 resource_group_name: str, 262 vm_scale_set_name: str, 263 **kwargs: Any 264 ) -> None: 265 cls = kwargs.pop('cls', None) # type: ClsType[None] 266 error_map = { 267 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 268 } 269 error_map.update(kwargs.pop('error_map', {})) 270 api_version = "2019-12-01" 271 272 # Construct URL 273 url = self._start_extension_upgrade_initial.metadata['url'] # type: ignore 274 path_format_arguments = { 275 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 276 'vmScaleSetName': self._serialize.url("vm_scale_set_name", vm_scale_set_name, 'str'), 277 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 278 } 279 url = self._client.format_url(url, **path_format_arguments) 280 281 # Construct parameters 282 query_parameters = {} # type: Dict[str, Any] 283 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 284 285 # Construct headers 286 header_parameters = {} # type: Dict[str, Any] 287 288 request = self._client.post(url, query_parameters, header_parameters) 289 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 290 response = pipeline_response.http_response 291 292 if response.status_code not in [200, 202]: 293 map_error(status_code=response.status_code, response=response, error_map=error_map) 294 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 295 296 if cls: 297 return cls(pipeline_response, None, {}) 298 299 _start_extension_upgrade_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachineScaleSets/{vmScaleSetName}/extensionRollingUpgrade'} # type: ignore 300 301 async def begin_start_extension_upgrade( 302 self, 303 resource_group_name: str, 304 vm_scale_set_name: str, 305 **kwargs: Any 306 ) -> AsyncLROPoller[None]: 307 """Starts a rolling upgrade to move all extensions for all virtual machine scale set instances to 308 the latest available extension version. Instances which are already running the latest 309 extension versions are not affected. 310 311 :param resource_group_name: The name of the resource group. 312 :type resource_group_name: str 313 :param vm_scale_set_name: The name of the VM scale set. 314 :type vm_scale_set_name: str 315 :keyword callable cls: A custom type or function that will be passed the direct response 316 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 317 :keyword polling: By default, your polling method will be AsyncARMPolling. 318 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 319 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 320 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 321 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 322 :rtype: ~azure.core.polling.AsyncLROPoller[None] 323 :raises ~azure.core.exceptions.HttpResponseError: 324 """ 325 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 326 cls = kwargs.pop('cls', None) # type: ClsType[None] 327 lro_delay = kwargs.pop( 328 'polling_interval', 329 self._config.polling_interval 330 ) 331 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 332 if cont_token is None: 333 raw_result = await self._start_extension_upgrade_initial( 334 resource_group_name=resource_group_name, 335 vm_scale_set_name=vm_scale_set_name, 336 cls=lambda x,y,z: x, 337 **kwargs 338 ) 339 340 kwargs.pop('error_map', None) 341 kwargs.pop('content_type', None) 342 343 def get_long_running_output(pipeline_response): 344 if cls: 345 return cls(pipeline_response, None, {}) 346 347 path_format_arguments = { 348 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 349 'vmScaleSetName': self._serialize.url("vm_scale_set_name", vm_scale_set_name, 'str'), 350 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 351 } 352 353 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 354 elif polling is False: polling_method = AsyncNoPolling() 355 else: polling_method = polling 356 if cont_token: 357 return AsyncLROPoller.from_continuation_token( 358 polling_method=polling_method, 359 continuation_token=cont_token, 360 client=self._client, 361 deserialization_callback=get_long_running_output 362 ) 363 else: 364 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 365 begin_start_extension_upgrade.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachineScaleSets/{vmScaleSetName}/extensionRollingUpgrade'} # type: ignore 366 367 async def get_latest( 368 self, 369 resource_group_name: str, 370 vm_scale_set_name: str, 371 **kwargs: Any 372 ) -> "_models.RollingUpgradeStatusInfo": 373 """Gets the status of the latest virtual machine scale set rolling upgrade. 374 375 :param resource_group_name: The name of the resource group. 376 :type resource_group_name: str 377 :param vm_scale_set_name: The name of the VM scale set. 378 :type vm_scale_set_name: str 379 :keyword callable cls: A custom type or function that will be passed the direct response 380 :return: RollingUpgradeStatusInfo, or the result of cls(response) 381 :rtype: ~azure.mgmt.compute.v2019_12_01.models.RollingUpgradeStatusInfo 382 :raises: ~azure.core.exceptions.HttpResponseError 383 """ 384 cls = kwargs.pop('cls', None) # type: ClsType["_models.RollingUpgradeStatusInfo"] 385 error_map = { 386 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 387 } 388 error_map.update(kwargs.pop('error_map', {})) 389 api_version = "2019-12-01" 390 accept = "application/json" 391 392 # Construct URL 393 url = self.get_latest.metadata['url'] # type: ignore 394 path_format_arguments = { 395 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 396 'vmScaleSetName': self._serialize.url("vm_scale_set_name", vm_scale_set_name, 'str'), 397 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 398 } 399 url = self._client.format_url(url, **path_format_arguments) 400 401 # Construct parameters 402 query_parameters = {} # type: Dict[str, Any] 403 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 404 405 # Construct headers 406 header_parameters = {} # type: Dict[str, Any] 407 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 408 409 request = self._client.get(url, query_parameters, header_parameters) 410 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 411 response = pipeline_response.http_response 412 413 if response.status_code not in [200]: 414 map_error(status_code=response.status_code, response=response, error_map=error_map) 415 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 416 417 deserialized = self._deserialize('RollingUpgradeStatusInfo', pipeline_response) 418 419 if cls: 420 return cls(pipeline_response, deserialized, {}) 421 422 return deserialized 423 get_latest.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/virtualMachineScaleSets/{vmScaleSetName}/rollingUpgrades/latest'} # type: ignore 424