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, Union 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.core.polling import AsyncLROPoller, AsyncNoPolling, AsyncPollingMethod 16from azure.mgmt.core.exceptions import ARMErrorFormat 17from azure.mgmt.core.polling.async_arm_polling import AsyncARMPolling 18 19from ... import models as _models 20 21T = TypeVar('T') 22ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]] 23 24class InboundNatRulesOperations: 25 """InboundNatRulesOperations async operations. 26 27 You should not instantiate this class directly. Instead, you should create a Client instance that 28 instantiates it for you and attaches it as an attribute. 29 30 :ivar models: Alias to model classes used in this operation group. 31 :type models: ~azure.mgmt.network.v2020_03_01.models 32 :param client: Client for service requests. 33 :param config: Configuration of service client. 34 :param serializer: An object model serializer. 35 :param deserializer: An object model deserializer. 36 """ 37 38 models = _models 39 40 def __init__(self, client, config, serializer, deserializer) -> None: 41 self._client = client 42 self._serialize = serializer 43 self._deserialize = deserializer 44 self._config = config 45 46 def list( 47 self, 48 resource_group_name: str, 49 load_balancer_name: str, 50 **kwargs 51 ) -> AsyncIterable["_models.InboundNatRuleListResult"]: 52 """Gets all the inbound nat rules in a load balancer. 53 54 :param resource_group_name: The name of the resource group. 55 :type resource_group_name: str 56 :param load_balancer_name: The name of the load balancer. 57 :type load_balancer_name: str 58 :keyword callable cls: A custom type or function that will be passed the direct response 59 :return: An iterator like instance of either InboundNatRuleListResult or the result of cls(response) 60 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_03_01.models.InboundNatRuleListResult] 61 :raises: ~azure.core.exceptions.HttpResponseError 62 """ 63 cls = kwargs.pop('cls', None) # type: ClsType["_models.InboundNatRuleListResult"] 64 error_map = { 65 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 66 } 67 error_map.update(kwargs.pop('error_map', {})) 68 api_version = "2020-03-01" 69 accept = "application/json" 70 71 def prepare_request(next_link=None): 72 # Construct headers 73 header_parameters = {} # type: Dict[str, Any] 74 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 75 76 if not next_link: 77 # Construct URL 78 url = self.list.metadata['url'] # type: ignore 79 path_format_arguments = { 80 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 81 'loadBalancerName': self._serialize.url("load_balancer_name", load_balancer_name, 'str'), 82 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 83 } 84 url = self._client.format_url(url, **path_format_arguments) 85 # Construct parameters 86 query_parameters = {} # type: Dict[str, Any] 87 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 88 89 request = self._client.get(url, query_parameters, header_parameters) 90 else: 91 url = next_link 92 query_parameters = {} # type: Dict[str, Any] 93 request = self._client.get(url, query_parameters, header_parameters) 94 return request 95 96 async def extract_data(pipeline_response): 97 deserialized = self._deserialize('InboundNatRuleListResult', pipeline_response) 98 list_of_elem = deserialized.value 99 if cls: 100 list_of_elem = cls(list_of_elem) 101 return deserialized.next_link or None, AsyncList(list_of_elem) 102 103 async def get_next(next_link=None): 104 request = prepare_request(next_link) 105 106 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 107 response = pipeline_response.http_response 108 109 if response.status_code not in [200]: 110 map_error(status_code=response.status_code, response=response, error_map=error_map) 111 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 112 113 return pipeline_response 114 115 return AsyncItemPaged( 116 get_next, extract_data 117 ) 118 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/loadBalancers/{loadBalancerName}/inboundNatRules'} # type: ignore 119 120 async def _delete_initial( 121 self, 122 resource_group_name: str, 123 load_balancer_name: str, 124 inbound_nat_rule_name: str, 125 **kwargs 126 ) -> None: 127 cls = kwargs.pop('cls', None) # type: ClsType[None] 128 error_map = { 129 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 130 } 131 error_map.update(kwargs.pop('error_map', {})) 132 api_version = "2020-03-01" 133 accept = "application/json" 134 135 # Construct URL 136 url = self._delete_initial.metadata['url'] # type: ignore 137 path_format_arguments = { 138 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 139 'loadBalancerName': self._serialize.url("load_balancer_name", load_balancer_name, 'str'), 140 'inboundNatRuleName': self._serialize.url("inbound_nat_rule_name", inbound_nat_rule_name, 'str'), 141 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 142 } 143 url = self._client.format_url(url, **path_format_arguments) 144 145 # Construct parameters 146 query_parameters = {} # type: Dict[str, Any] 147 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 148 149 # Construct headers 150 header_parameters = {} # type: Dict[str, Any] 151 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 152 153 request = self._client.delete(url, query_parameters, header_parameters) 154 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 155 response = pipeline_response.http_response 156 157 if response.status_code not in [200, 202, 204]: 158 map_error(status_code=response.status_code, response=response, error_map=error_map) 159 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 160 161 if cls: 162 return cls(pipeline_response, None, {}) 163 164 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/loadBalancers/{loadBalancerName}/inboundNatRules/{inboundNatRuleName}'} # type: ignore 165 166 async def begin_delete( 167 self, 168 resource_group_name: str, 169 load_balancer_name: str, 170 inbound_nat_rule_name: str, 171 **kwargs 172 ) -> AsyncLROPoller[None]: 173 """Deletes the specified load balancer inbound nat rule. 174 175 :param resource_group_name: The name of the resource group. 176 :type resource_group_name: str 177 :param load_balancer_name: The name of the load balancer. 178 :type load_balancer_name: str 179 :param inbound_nat_rule_name: The name of the inbound nat rule. 180 :type inbound_nat_rule_name: str 181 :keyword callable cls: A custom type or function that will be passed the direct response 182 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 183 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 184 False for no polling, or your own initialized polling object for a personal polling strategy. 185 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 186 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 187 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 188 :rtype: ~azure.core.polling.AsyncLROPoller[None] 189 :raises ~azure.core.exceptions.HttpResponseError: 190 """ 191 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 192 cls = kwargs.pop('cls', None) # type: ClsType[None] 193 lro_delay = kwargs.pop( 194 'polling_interval', 195 self._config.polling_interval 196 ) 197 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 198 if cont_token is None: 199 raw_result = await self._delete_initial( 200 resource_group_name=resource_group_name, 201 load_balancer_name=load_balancer_name, 202 inbound_nat_rule_name=inbound_nat_rule_name, 203 cls=lambda x,y,z: x, 204 **kwargs 205 ) 206 207 kwargs.pop('error_map', None) 208 kwargs.pop('content_type', None) 209 210 def get_long_running_output(pipeline_response): 211 if cls: 212 return cls(pipeline_response, None, {}) 213 214 path_format_arguments = { 215 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 216 'loadBalancerName': self._serialize.url("load_balancer_name", load_balancer_name, 'str'), 217 'inboundNatRuleName': self._serialize.url("inbound_nat_rule_name", inbound_nat_rule_name, 'str'), 218 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 219 } 220 221 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 222 elif polling is False: polling_method = AsyncNoPolling() 223 else: polling_method = polling 224 if cont_token: 225 return AsyncLROPoller.from_continuation_token( 226 polling_method=polling_method, 227 continuation_token=cont_token, 228 client=self._client, 229 deserialization_callback=get_long_running_output 230 ) 231 else: 232 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 233 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/loadBalancers/{loadBalancerName}/inboundNatRules/{inboundNatRuleName}'} # type: ignore 234 235 async def get( 236 self, 237 resource_group_name: str, 238 load_balancer_name: str, 239 inbound_nat_rule_name: str, 240 expand: Optional[str] = None, 241 **kwargs 242 ) -> "_models.InboundNatRule": 243 """Gets the specified load balancer inbound nat rule. 244 245 :param resource_group_name: The name of the resource group. 246 :type resource_group_name: str 247 :param load_balancer_name: The name of the load balancer. 248 :type load_balancer_name: str 249 :param inbound_nat_rule_name: The name of the inbound nat rule. 250 :type inbound_nat_rule_name: str 251 :param expand: Expands referenced resources. 252 :type expand: str 253 :keyword callable cls: A custom type or function that will be passed the direct response 254 :return: InboundNatRule, or the result of cls(response) 255 :rtype: ~azure.mgmt.network.v2020_03_01.models.InboundNatRule 256 :raises: ~azure.core.exceptions.HttpResponseError 257 """ 258 cls = kwargs.pop('cls', None) # type: ClsType["_models.InboundNatRule"] 259 error_map = { 260 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 261 } 262 error_map.update(kwargs.pop('error_map', {})) 263 api_version = "2020-03-01" 264 accept = "application/json" 265 266 # Construct URL 267 url = self.get.metadata['url'] # type: ignore 268 path_format_arguments = { 269 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 270 'loadBalancerName': self._serialize.url("load_balancer_name", load_balancer_name, 'str'), 271 'inboundNatRuleName': self._serialize.url("inbound_nat_rule_name", inbound_nat_rule_name, 'str'), 272 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 273 } 274 url = self._client.format_url(url, **path_format_arguments) 275 276 # Construct parameters 277 query_parameters = {} # type: Dict[str, Any] 278 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 279 if expand is not None: 280 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 281 282 # Construct headers 283 header_parameters = {} # type: Dict[str, Any] 284 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 285 286 request = self._client.get(url, query_parameters, header_parameters) 287 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 288 response = pipeline_response.http_response 289 290 if response.status_code not in [200]: 291 map_error(status_code=response.status_code, response=response, error_map=error_map) 292 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 293 294 deserialized = self._deserialize('InboundNatRule', pipeline_response) 295 296 if cls: 297 return cls(pipeline_response, deserialized, {}) 298 299 return deserialized 300 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/loadBalancers/{loadBalancerName}/inboundNatRules/{inboundNatRuleName}'} # type: ignore 301 302 async def _create_or_update_initial( 303 self, 304 resource_group_name: str, 305 load_balancer_name: str, 306 inbound_nat_rule_name: str, 307 inbound_nat_rule_parameters: "_models.InboundNatRule", 308 **kwargs 309 ) -> "_models.InboundNatRule": 310 cls = kwargs.pop('cls', None) # type: ClsType["_models.InboundNatRule"] 311 error_map = { 312 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 313 } 314 error_map.update(kwargs.pop('error_map', {})) 315 api_version = "2020-03-01" 316 content_type = kwargs.pop("content_type", "application/json") 317 accept = "application/json" 318 319 # Construct URL 320 url = self._create_or_update_initial.metadata['url'] # type: ignore 321 path_format_arguments = { 322 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 323 'loadBalancerName': self._serialize.url("load_balancer_name", load_balancer_name, 'str'), 324 'inboundNatRuleName': self._serialize.url("inbound_nat_rule_name", inbound_nat_rule_name, 'str'), 325 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 326 } 327 url = self._client.format_url(url, **path_format_arguments) 328 329 # Construct parameters 330 query_parameters = {} # type: Dict[str, Any] 331 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 332 333 # Construct headers 334 header_parameters = {} # type: Dict[str, Any] 335 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 336 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 337 338 body_content_kwargs = {} # type: Dict[str, Any] 339 body_content = self._serialize.body(inbound_nat_rule_parameters, 'InboundNatRule') 340 body_content_kwargs['content'] = body_content 341 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 342 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 343 response = pipeline_response.http_response 344 345 if response.status_code not in [200, 201]: 346 map_error(status_code=response.status_code, response=response, error_map=error_map) 347 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 348 349 if response.status_code == 200: 350 deserialized = self._deserialize('InboundNatRule', pipeline_response) 351 352 if response.status_code == 201: 353 deserialized = self._deserialize('InboundNatRule', pipeline_response) 354 355 if cls: 356 return cls(pipeline_response, deserialized, {}) 357 358 return deserialized 359 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/loadBalancers/{loadBalancerName}/inboundNatRules/{inboundNatRuleName}'} # type: ignore 360 361 async def begin_create_or_update( 362 self, 363 resource_group_name: str, 364 load_balancer_name: str, 365 inbound_nat_rule_name: str, 366 inbound_nat_rule_parameters: "_models.InboundNatRule", 367 **kwargs 368 ) -> AsyncLROPoller["_models.InboundNatRule"]: 369 """Creates or updates a load balancer inbound nat rule. 370 371 :param resource_group_name: The name of the resource group. 372 :type resource_group_name: str 373 :param load_balancer_name: The name of the load balancer. 374 :type load_balancer_name: str 375 :param inbound_nat_rule_name: The name of the inbound nat rule. 376 :type inbound_nat_rule_name: str 377 :param inbound_nat_rule_parameters: Parameters supplied to the create or update inbound nat 378 rule operation. 379 :type inbound_nat_rule_parameters: ~azure.mgmt.network.v2020_03_01.models.InboundNatRule 380 :keyword callable cls: A custom type or function that will be passed the direct response 381 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 382 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 383 False for no polling, or your own initialized polling object for a personal polling strategy. 384 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 385 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 386 :return: An instance of AsyncLROPoller that returns either InboundNatRule or the result of cls(response) 387 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_03_01.models.InboundNatRule] 388 :raises ~azure.core.exceptions.HttpResponseError: 389 """ 390 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 391 cls = kwargs.pop('cls', None) # type: ClsType["_models.InboundNatRule"] 392 lro_delay = kwargs.pop( 393 'polling_interval', 394 self._config.polling_interval 395 ) 396 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 397 if cont_token is None: 398 raw_result = await self._create_or_update_initial( 399 resource_group_name=resource_group_name, 400 load_balancer_name=load_balancer_name, 401 inbound_nat_rule_name=inbound_nat_rule_name, 402 inbound_nat_rule_parameters=inbound_nat_rule_parameters, 403 cls=lambda x,y,z: x, 404 **kwargs 405 ) 406 407 kwargs.pop('error_map', None) 408 kwargs.pop('content_type', None) 409 410 def get_long_running_output(pipeline_response): 411 deserialized = self._deserialize('InboundNatRule', pipeline_response) 412 413 if cls: 414 return cls(pipeline_response, deserialized, {}) 415 return deserialized 416 417 path_format_arguments = { 418 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 419 'loadBalancerName': self._serialize.url("load_balancer_name", load_balancer_name, 'str'), 420 'inboundNatRuleName': self._serialize.url("inbound_nat_rule_name", inbound_nat_rule_name, 'str'), 421 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 422 } 423 424 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 425 elif polling is False: polling_method = AsyncNoPolling() 426 else: polling_method = polling 427 if cont_token: 428 return AsyncLROPoller.from_continuation_token( 429 polling_method=polling_method, 430 continuation_token=cont_token, 431 client=self._client, 432 deserialization_callback=get_long_running_output 433 ) 434 else: 435 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 436 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/loadBalancers/{loadBalancerName}/inboundNatRules/{inboundNatRuleName}'} # type: ignore 437