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, AsyncIterable, Callable, Dict, Generic, Optional, TypeVar 9import warnings 10 11from azure.core.async_paging import AsyncItemPaged, AsyncList 12from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error 13from azure.core.pipeline import PipelineResponse 14from azure.core.pipeline.transport import AsyncHttpResponse, HttpRequest 15from azure.mgmt.core.exceptions import ARMErrorFormat 16 17from ... import models as _models 18 19T = TypeVar('T') 20ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]] 21 22class AvailabilitySetsOperations: 23 """AvailabilitySetsOperations async operations. 24 25 You should not instantiate this class directly. Instead, you should create a Client instance that 26 instantiates it for you and attaches it as an attribute. 27 28 :ivar models: Alias to model classes used in this operation group. 29 :type models: ~azure.mgmt.compute.v2018_06_01.models 30 :param client: Client for service requests. 31 :param config: Configuration of service client. 32 :param serializer: An object model serializer. 33 :param deserializer: An object model deserializer. 34 """ 35 36 models = _models 37 38 def __init__(self, client, config, serializer, deserializer) -> None: 39 self._client = client 40 self._serialize = serializer 41 self._deserialize = deserializer 42 self._config = config 43 44 async def create_or_update( 45 self, 46 resource_group_name: str, 47 availability_set_name: str, 48 parameters: "_models.AvailabilitySet", 49 **kwargs: Any 50 ) -> "_models.AvailabilitySet": 51 """Create or update an availability set. 52 53 :param resource_group_name: The name of the resource group. 54 :type resource_group_name: str 55 :param availability_set_name: The name of the availability set. 56 :type availability_set_name: str 57 :param parameters: Parameters supplied to the Create Availability Set operation. 58 :type parameters: ~azure.mgmt.compute.v2018_06_01.models.AvailabilitySet 59 :keyword callable cls: A custom type or function that will be passed the direct response 60 :return: AvailabilitySet, or the result of cls(response) 61 :rtype: ~azure.mgmt.compute.v2018_06_01.models.AvailabilitySet 62 :raises: ~azure.core.exceptions.HttpResponseError 63 """ 64 cls = kwargs.pop('cls', None) # type: ClsType["_models.AvailabilitySet"] 65 error_map = { 66 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 67 } 68 error_map.update(kwargs.pop('error_map', {})) 69 api_version = "2018-06-01" 70 content_type = kwargs.pop("content_type", "application/json") 71 accept = "application/json" 72 73 # Construct URL 74 url = self.create_or_update.metadata['url'] # type: ignore 75 path_format_arguments = { 76 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 77 'availabilitySetName': self._serialize.url("availability_set_name", availability_set_name, 'str'), 78 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 79 } 80 url = self._client.format_url(url, **path_format_arguments) 81 82 # Construct parameters 83 query_parameters = {} # type: Dict[str, Any] 84 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 85 86 # Construct headers 87 header_parameters = {} # type: Dict[str, Any] 88 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 89 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 90 91 body_content_kwargs = {} # type: Dict[str, Any] 92 body_content = self._serialize.body(parameters, 'AvailabilitySet') 93 body_content_kwargs['content'] = body_content 94 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 95 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 96 response = pipeline_response.http_response 97 98 if response.status_code not in [200]: 99 map_error(status_code=response.status_code, response=response, error_map=error_map) 100 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 101 102 deserialized = self._deserialize('AvailabilitySet', pipeline_response) 103 104 if cls: 105 return cls(pipeline_response, deserialized, {}) 106 107 return deserialized 108 create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/availabilitySets/{availabilitySetName}'} # type: ignore 109 110 async def update( 111 self, 112 resource_group_name: str, 113 availability_set_name: str, 114 parameters: "_models.AvailabilitySetUpdate", 115 **kwargs: Any 116 ) -> "_models.AvailabilitySet": 117 """Update an availability set. 118 119 :param resource_group_name: The name of the resource group. 120 :type resource_group_name: str 121 :param availability_set_name: The name of the availability set. 122 :type availability_set_name: str 123 :param parameters: Parameters supplied to the Update Availability Set operation. 124 :type parameters: ~azure.mgmt.compute.v2018_06_01.models.AvailabilitySetUpdate 125 :keyword callable cls: A custom type or function that will be passed the direct response 126 :return: AvailabilitySet, or the result of cls(response) 127 :rtype: ~azure.mgmt.compute.v2018_06_01.models.AvailabilitySet 128 :raises: ~azure.core.exceptions.HttpResponseError 129 """ 130 cls = kwargs.pop('cls', None) # type: ClsType["_models.AvailabilitySet"] 131 error_map = { 132 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 133 } 134 error_map.update(kwargs.pop('error_map', {})) 135 api_version = "2018-06-01" 136 content_type = kwargs.pop("content_type", "application/json") 137 accept = "application/json" 138 139 # Construct URL 140 url = self.update.metadata['url'] # type: ignore 141 path_format_arguments = { 142 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 143 'availabilitySetName': self._serialize.url("availability_set_name", availability_set_name, 'str'), 144 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 145 } 146 url = self._client.format_url(url, **path_format_arguments) 147 148 # Construct parameters 149 query_parameters = {} # type: Dict[str, Any] 150 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 151 152 # Construct headers 153 header_parameters = {} # type: Dict[str, Any] 154 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 155 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 156 157 body_content_kwargs = {} # type: Dict[str, Any] 158 body_content = self._serialize.body(parameters, 'AvailabilitySetUpdate') 159 body_content_kwargs['content'] = body_content 160 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 161 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 162 response = pipeline_response.http_response 163 164 if response.status_code not in [200]: 165 map_error(status_code=response.status_code, response=response, error_map=error_map) 166 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 167 168 deserialized = self._deserialize('AvailabilitySet', pipeline_response) 169 170 if cls: 171 return cls(pipeline_response, deserialized, {}) 172 173 return deserialized 174 update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/availabilitySets/{availabilitySetName}'} # type: ignore 175 176 async def delete( 177 self, 178 resource_group_name: str, 179 availability_set_name: str, 180 **kwargs: Any 181 ) -> None: 182 """Delete an availability set. 183 184 :param resource_group_name: The name of the resource group. 185 :type resource_group_name: str 186 :param availability_set_name: The name of the availability set. 187 :type availability_set_name: str 188 :keyword callable cls: A custom type or function that will be passed the direct response 189 :return: None, or the result of cls(response) 190 :rtype: None 191 :raises: ~azure.core.exceptions.HttpResponseError 192 """ 193 cls = kwargs.pop('cls', None) # type: ClsType[None] 194 error_map = { 195 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 196 } 197 error_map.update(kwargs.pop('error_map', {})) 198 api_version = "2018-06-01" 199 200 # Construct URL 201 url = self.delete.metadata['url'] # type: ignore 202 path_format_arguments = { 203 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 204 'availabilitySetName': self._serialize.url("availability_set_name", availability_set_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 216 request = self._client.delete(url, query_parameters, header_parameters) 217 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 218 response = pipeline_response.http_response 219 220 if response.status_code not in [200, 204]: 221 map_error(status_code=response.status_code, response=response, error_map=error_map) 222 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 223 224 if cls: 225 return cls(pipeline_response, None, {}) 226 227 delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/availabilitySets/{availabilitySetName}'} # type: ignore 228 229 async def get( 230 self, 231 resource_group_name: str, 232 availability_set_name: str, 233 **kwargs: Any 234 ) -> "_models.AvailabilitySet": 235 """Retrieves information about an availability set. 236 237 :param resource_group_name: The name of the resource group. 238 :type resource_group_name: str 239 :param availability_set_name: The name of the availability set. 240 :type availability_set_name: str 241 :keyword callable cls: A custom type or function that will be passed the direct response 242 :return: AvailabilitySet, or the result of cls(response) 243 :rtype: ~azure.mgmt.compute.v2018_06_01.models.AvailabilitySet 244 :raises: ~azure.core.exceptions.HttpResponseError 245 """ 246 cls = kwargs.pop('cls', None) # type: ClsType["_models.AvailabilitySet"] 247 error_map = { 248 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 249 } 250 error_map.update(kwargs.pop('error_map', {})) 251 api_version = "2018-06-01" 252 accept = "application/json" 253 254 # Construct URL 255 url = self.get.metadata['url'] # type: ignore 256 path_format_arguments = { 257 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 258 'availabilitySetName': self._serialize.url("availability_set_name", availability_set_name, 'str'), 259 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 260 } 261 url = self._client.format_url(url, **path_format_arguments) 262 263 # Construct parameters 264 query_parameters = {} # type: Dict[str, Any] 265 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 266 267 # Construct headers 268 header_parameters = {} # type: Dict[str, Any] 269 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 270 271 request = self._client.get(url, query_parameters, header_parameters) 272 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 273 response = pipeline_response.http_response 274 275 if response.status_code not in [200]: 276 map_error(status_code=response.status_code, response=response, error_map=error_map) 277 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 278 279 deserialized = self._deserialize('AvailabilitySet', pipeline_response) 280 281 if cls: 282 return cls(pipeline_response, deserialized, {}) 283 284 return deserialized 285 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/availabilitySets/{availabilitySetName}'} # type: ignore 286 287 def list_by_subscription( 288 self, 289 expand: Optional[str] = None, 290 **kwargs: Any 291 ) -> AsyncIterable["_models.AvailabilitySetListResult"]: 292 """Lists all availability sets in a subscription. 293 294 :param expand: The expand expression to apply to the operation. Allowed values are 295 'instanceView'. 296 :type expand: str 297 :keyword callable cls: A custom type or function that will be passed the direct response 298 :return: An iterator like instance of either AvailabilitySetListResult or the result of cls(response) 299 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.compute.v2018_06_01.models.AvailabilitySetListResult] 300 :raises: ~azure.core.exceptions.HttpResponseError 301 """ 302 cls = kwargs.pop('cls', None) # type: ClsType["_models.AvailabilitySetListResult"] 303 error_map = { 304 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 305 } 306 error_map.update(kwargs.pop('error_map', {})) 307 api_version = "2018-06-01" 308 accept = "application/json" 309 310 def prepare_request(next_link=None): 311 # Construct headers 312 header_parameters = {} # type: Dict[str, Any] 313 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 314 315 if not next_link: 316 # Construct URL 317 url = self.list_by_subscription.metadata['url'] # type: ignore 318 path_format_arguments = { 319 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 320 } 321 url = self._client.format_url(url, **path_format_arguments) 322 # Construct parameters 323 query_parameters = {} # type: Dict[str, Any] 324 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 325 if expand is not None: 326 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 327 328 request = self._client.get(url, query_parameters, header_parameters) 329 else: 330 url = next_link 331 query_parameters = {} # type: Dict[str, Any] 332 request = self._client.get(url, query_parameters, header_parameters) 333 return request 334 335 async def extract_data(pipeline_response): 336 deserialized = self._deserialize('AvailabilitySetListResult', pipeline_response) 337 list_of_elem = deserialized.value 338 if cls: 339 list_of_elem = cls(list_of_elem) 340 return deserialized.next_link or None, AsyncList(list_of_elem) 341 342 async def get_next(next_link=None): 343 request = prepare_request(next_link) 344 345 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 346 response = pipeline_response.http_response 347 348 if response.status_code not in [200]: 349 map_error(status_code=response.status_code, response=response, error_map=error_map) 350 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 351 352 return pipeline_response 353 354 return AsyncItemPaged( 355 get_next, extract_data 356 ) 357 list_by_subscription.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Compute/availabilitySets'} # type: ignore 358 359 def list( 360 self, 361 resource_group_name: str, 362 **kwargs: Any 363 ) -> AsyncIterable["_models.AvailabilitySetListResult"]: 364 """Lists all availability sets in a resource group. 365 366 :param resource_group_name: The name of the resource group. 367 :type resource_group_name: str 368 :keyword callable cls: A custom type or function that will be passed the direct response 369 :return: An iterator like instance of either AvailabilitySetListResult or the result of cls(response) 370 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.compute.v2018_06_01.models.AvailabilitySetListResult] 371 :raises: ~azure.core.exceptions.HttpResponseError 372 """ 373 cls = kwargs.pop('cls', None) # type: ClsType["_models.AvailabilitySetListResult"] 374 error_map = { 375 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 376 } 377 error_map.update(kwargs.pop('error_map', {})) 378 api_version = "2018-06-01" 379 accept = "application/json" 380 381 def prepare_request(next_link=None): 382 # Construct headers 383 header_parameters = {} # type: Dict[str, Any] 384 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 385 386 if not next_link: 387 # Construct URL 388 url = self.list.metadata['url'] # type: ignore 389 path_format_arguments = { 390 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 391 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 392 } 393 url = self._client.format_url(url, **path_format_arguments) 394 # Construct parameters 395 query_parameters = {} # type: Dict[str, Any] 396 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 397 398 request = self._client.get(url, query_parameters, header_parameters) 399 else: 400 url = next_link 401 query_parameters = {} # type: Dict[str, Any] 402 request = self._client.get(url, query_parameters, header_parameters) 403 return request 404 405 async def extract_data(pipeline_response): 406 deserialized = self._deserialize('AvailabilitySetListResult', pipeline_response) 407 list_of_elem = deserialized.value 408 if cls: 409 list_of_elem = cls(list_of_elem) 410 return deserialized.next_link or None, AsyncList(list_of_elem) 411 412 async def get_next(next_link=None): 413 request = prepare_request(next_link) 414 415 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 416 response = pipeline_response.http_response 417 418 if response.status_code not in [200]: 419 map_error(status_code=response.status_code, response=response, error_map=error_map) 420 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 421 422 return pipeline_response 423 424 return AsyncItemPaged( 425 get_next, extract_data 426 ) 427 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/availabilitySets'} # type: ignore 428 429 def list_available_sizes( 430 self, 431 resource_group_name: str, 432 availability_set_name: str, 433 **kwargs: Any 434 ) -> AsyncIterable["_models.VirtualMachineSizeListResult"]: 435 """Lists all available virtual machine sizes that can be used to create a new virtual machine in 436 an existing availability set. 437 438 :param resource_group_name: The name of the resource group. 439 :type resource_group_name: str 440 :param availability_set_name: The name of the availability set. 441 :type availability_set_name: str 442 :keyword callable cls: A custom type or function that will be passed the direct response 443 :return: An iterator like instance of either VirtualMachineSizeListResult or the result of cls(response) 444 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.compute.v2018_06_01.models.VirtualMachineSizeListResult] 445 :raises: ~azure.core.exceptions.HttpResponseError 446 """ 447 cls = kwargs.pop('cls', None) # type: ClsType["_models.VirtualMachineSizeListResult"] 448 error_map = { 449 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 450 } 451 error_map.update(kwargs.pop('error_map', {})) 452 api_version = "2018-06-01" 453 accept = "application/json" 454 455 def prepare_request(next_link=None): 456 # Construct headers 457 header_parameters = {} # type: Dict[str, Any] 458 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 459 460 if not next_link: 461 # Construct URL 462 url = self.list_available_sizes.metadata['url'] # type: ignore 463 path_format_arguments = { 464 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 465 'availabilitySetName': self._serialize.url("availability_set_name", availability_set_name, 'str'), 466 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 467 } 468 url = self._client.format_url(url, **path_format_arguments) 469 # Construct parameters 470 query_parameters = {} # type: Dict[str, Any] 471 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 472 473 request = self._client.get(url, query_parameters, header_parameters) 474 else: 475 url = next_link 476 query_parameters = {} # type: Dict[str, Any] 477 request = self._client.get(url, query_parameters, header_parameters) 478 return request 479 480 async def extract_data(pipeline_response): 481 deserialized = self._deserialize('VirtualMachineSizeListResult', pipeline_response) 482 list_of_elem = deserialized.value 483 if cls: 484 list_of_elem = cls(list_of_elem) 485 return None, AsyncList(list_of_elem) 486 487 async def get_next(next_link=None): 488 request = prepare_request(next_link) 489 490 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 491 response = pipeline_response.http_response 492 493 if response.status_code not in [200]: 494 map_error(status_code=response.status_code, response=response, error_map=error_map) 495 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 496 497 return pipeline_response 498 499 return AsyncItemPaged( 500 get_next, extract_data 501 ) 502 list_available_sizes.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/availabilitySets/{availabilitySetName}/vmSizes'} # type: ignore 503