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