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 ApplicationGatewaysOperations: 25 """ApplicationGatewaysOperations 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.v2017_10_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 async def _delete_initial( 47 self, 48 resource_group_name: str, 49 application_gateway_name: str, 50 **kwargs 51 ) -> None: 52 cls = kwargs.pop('cls', None) # type: ClsType[None] 53 error_map = { 54 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 55 } 56 error_map.update(kwargs.pop('error_map', {})) 57 api_version = "2017-10-01" 58 59 # Construct URL 60 url = self._delete_initial.metadata['url'] # type: ignore 61 path_format_arguments = { 62 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 63 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 64 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 65 } 66 url = self._client.format_url(url, **path_format_arguments) 67 68 # Construct parameters 69 query_parameters = {} # type: Dict[str, Any] 70 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 71 72 # Construct headers 73 header_parameters = {} # type: Dict[str, Any] 74 75 request = self._client.delete(url, query_parameters, header_parameters) 76 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 77 response = pipeline_response.http_response 78 79 if response.status_code not in [200, 202, 204]: 80 map_error(status_code=response.status_code, response=response, error_map=error_map) 81 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 82 83 if cls: 84 return cls(pipeline_response, None, {}) 85 86 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}'} # type: ignore 87 88 async def begin_delete( 89 self, 90 resource_group_name: str, 91 application_gateway_name: str, 92 **kwargs 93 ) -> AsyncLROPoller[None]: 94 """Deletes the specified application gateway. 95 96 :param resource_group_name: The name of the resource group. 97 :type resource_group_name: str 98 :param application_gateway_name: The name of the application gateway. 99 :type application_gateway_name: str 100 :keyword callable cls: A custom type or function that will be passed the direct response 101 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 102 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 103 False for no polling, or your own initialized polling object for a personal polling strategy. 104 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 105 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 106 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 107 :rtype: ~azure.core.polling.AsyncLROPoller[None] 108 :raises ~azure.core.exceptions.HttpResponseError: 109 """ 110 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 111 cls = kwargs.pop('cls', None) # type: ClsType[None] 112 lro_delay = kwargs.pop( 113 'polling_interval', 114 self._config.polling_interval 115 ) 116 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 117 if cont_token is None: 118 raw_result = await self._delete_initial( 119 resource_group_name=resource_group_name, 120 application_gateway_name=application_gateway_name, 121 cls=lambda x,y,z: x, 122 **kwargs 123 ) 124 125 kwargs.pop('error_map', None) 126 kwargs.pop('content_type', None) 127 128 def get_long_running_output(pipeline_response): 129 if cls: 130 return cls(pipeline_response, None, {}) 131 132 path_format_arguments = { 133 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 134 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 135 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 136 } 137 138 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 139 elif polling is False: polling_method = AsyncNoPolling() 140 else: polling_method = polling 141 if cont_token: 142 return AsyncLROPoller.from_continuation_token( 143 polling_method=polling_method, 144 continuation_token=cont_token, 145 client=self._client, 146 deserialization_callback=get_long_running_output 147 ) 148 else: 149 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 150 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}'} # type: ignore 151 152 async def get( 153 self, 154 resource_group_name: str, 155 application_gateway_name: str, 156 **kwargs 157 ) -> "_models.ApplicationGateway": 158 """Gets the specified application gateway. 159 160 :param resource_group_name: The name of the resource group. 161 :type resource_group_name: str 162 :param application_gateway_name: The name of the application gateway. 163 :type application_gateway_name: str 164 :keyword callable cls: A custom type or function that will be passed the direct response 165 :return: ApplicationGateway, or the result of cls(response) 166 :rtype: ~azure.mgmt.network.v2017_10_01.models.ApplicationGateway 167 :raises: ~azure.core.exceptions.HttpResponseError 168 """ 169 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGateway"] 170 error_map = { 171 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 172 } 173 error_map.update(kwargs.pop('error_map', {})) 174 api_version = "2017-10-01" 175 accept = "application/json" 176 177 # Construct URL 178 url = self.get.metadata['url'] # type: ignore 179 path_format_arguments = { 180 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 181 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 182 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 183 } 184 url = self._client.format_url(url, **path_format_arguments) 185 186 # Construct parameters 187 query_parameters = {} # type: Dict[str, Any] 188 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 189 190 # Construct headers 191 header_parameters = {} # type: Dict[str, Any] 192 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 193 194 request = self._client.get(url, query_parameters, header_parameters) 195 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 196 response = pipeline_response.http_response 197 198 if response.status_code not in [200]: 199 map_error(status_code=response.status_code, response=response, error_map=error_map) 200 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 201 202 deserialized = self._deserialize('ApplicationGateway', pipeline_response) 203 204 if cls: 205 return cls(pipeline_response, deserialized, {}) 206 207 return deserialized 208 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}'} # type: ignore 209 210 async def _create_or_update_initial( 211 self, 212 resource_group_name: str, 213 application_gateway_name: str, 214 parameters: "_models.ApplicationGateway", 215 **kwargs 216 ) -> "_models.ApplicationGateway": 217 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGateway"] 218 error_map = { 219 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 220 } 221 error_map.update(kwargs.pop('error_map', {})) 222 api_version = "2017-10-01" 223 content_type = kwargs.pop("content_type", "application/json") 224 accept = "application/json" 225 226 # Construct URL 227 url = self._create_or_update_initial.metadata['url'] # type: ignore 228 path_format_arguments = { 229 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 230 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 231 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 232 } 233 url = self._client.format_url(url, **path_format_arguments) 234 235 # Construct parameters 236 query_parameters = {} # type: Dict[str, Any] 237 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 238 239 # Construct headers 240 header_parameters = {} # type: Dict[str, Any] 241 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 242 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 243 244 body_content_kwargs = {} # type: Dict[str, Any] 245 body_content = self._serialize.body(parameters, 'ApplicationGateway') 246 body_content_kwargs['content'] = body_content 247 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 248 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 249 response = pipeline_response.http_response 250 251 if response.status_code not in [200, 201]: 252 map_error(status_code=response.status_code, response=response, error_map=error_map) 253 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 254 255 if response.status_code == 200: 256 deserialized = self._deserialize('ApplicationGateway', pipeline_response) 257 258 if response.status_code == 201: 259 deserialized = self._deserialize('ApplicationGateway', pipeline_response) 260 261 if cls: 262 return cls(pipeline_response, deserialized, {}) 263 264 return deserialized 265 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}'} # type: ignore 266 267 async def begin_create_or_update( 268 self, 269 resource_group_name: str, 270 application_gateway_name: str, 271 parameters: "_models.ApplicationGateway", 272 **kwargs 273 ) -> AsyncLROPoller["_models.ApplicationGateway"]: 274 """Creates or updates the specified application gateway. 275 276 :param resource_group_name: The name of the resource group. 277 :type resource_group_name: str 278 :param application_gateway_name: The name of the application gateway. 279 :type application_gateway_name: str 280 :param parameters: Parameters supplied to the create or update application gateway operation. 281 :type parameters: ~azure.mgmt.network.v2017_10_01.models.ApplicationGateway 282 :keyword callable cls: A custom type or function that will be passed the direct response 283 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 284 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 285 False for no polling, or your own initialized polling object for a personal polling strategy. 286 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 287 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 288 :return: An instance of AsyncLROPoller that returns either ApplicationGateway or the result of cls(response) 289 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2017_10_01.models.ApplicationGateway] 290 :raises ~azure.core.exceptions.HttpResponseError: 291 """ 292 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 293 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGateway"] 294 lro_delay = kwargs.pop( 295 'polling_interval', 296 self._config.polling_interval 297 ) 298 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 299 if cont_token is None: 300 raw_result = await self._create_or_update_initial( 301 resource_group_name=resource_group_name, 302 application_gateway_name=application_gateway_name, 303 parameters=parameters, 304 cls=lambda x,y,z: x, 305 **kwargs 306 ) 307 308 kwargs.pop('error_map', None) 309 kwargs.pop('content_type', None) 310 311 def get_long_running_output(pipeline_response): 312 deserialized = self._deserialize('ApplicationGateway', pipeline_response) 313 314 if cls: 315 return cls(pipeline_response, deserialized, {}) 316 return deserialized 317 318 path_format_arguments = { 319 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 320 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 321 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 322 } 323 324 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 325 elif polling is False: polling_method = AsyncNoPolling() 326 else: polling_method = polling 327 if cont_token: 328 return AsyncLROPoller.from_continuation_token( 329 polling_method=polling_method, 330 continuation_token=cont_token, 331 client=self._client, 332 deserialization_callback=get_long_running_output 333 ) 334 else: 335 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 336 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}'} # type: ignore 337 338 async def _update_tags_initial( 339 self, 340 resource_group_name: str, 341 application_gateway_name: str, 342 parameters: "_models.TagsObject", 343 **kwargs 344 ) -> "_models.ApplicationGateway": 345 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGateway"] 346 error_map = { 347 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 348 } 349 error_map.update(kwargs.pop('error_map', {})) 350 api_version = "2017-10-01" 351 content_type = kwargs.pop("content_type", "application/json") 352 accept = "application/json" 353 354 # Construct URL 355 url = self._update_tags_initial.metadata['url'] # type: ignore 356 path_format_arguments = { 357 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 358 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 359 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 360 } 361 url = self._client.format_url(url, **path_format_arguments) 362 363 # Construct parameters 364 query_parameters = {} # type: Dict[str, Any] 365 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 366 367 # Construct headers 368 header_parameters = {} # type: Dict[str, Any] 369 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 370 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 371 372 body_content_kwargs = {} # type: Dict[str, Any] 373 body_content = self._serialize.body(parameters, 'TagsObject') 374 body_content_kwargs['content'] = body_content 375 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 376 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 377 response = pipeline_response.http_response 378 379 if response.status_code not in [200]: 380 map_error(status_code=response.status_code, response=response, error_map=error_map) 381 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 382 383 deserialized = self._deserialize('ApplicationGateway', pipeline_response) 384 385 if cls: 386 return cls(pipeline_response, deserialized, {}) 387 388 return deserialized 389 _update_tags_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}'} # type: ignore 390 391 async def begin_update_tags( 392 self, 393 resource_group_name: str, 394 application_gateway_name: str, 395 parameters: "_models.TagsObject", 396 **kwargs 397 ) -> AsyncLROPoller["_models.ApplicationGateway"]: 398 """Updates the specified application gateway tags. 399 400 :param resource_group_name: The name of the resource group. 401 :type resource_group_name: str 402 :param application_gateway_name: The name of the application gateway. 403 :type application_gateway_name: str 404 :param parameters: Parameters supplied to update application gateway tags. 405 :type parameters: ~azure.mgmt.network.v2017_10_01.models.TagsObject 406 :keyword callable cls: A custom type or function that will be passed the direct response 407 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 408 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 409 False for no polling, or your own initialized polling object for a personal polling strategy. 410 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 411 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 412 :return: An instance of AsyncLROPoller that returns either ApplicationGateway or the result of cls(response) 413 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2017_10_01.models.ApplicationGateway] 414 :raises ~azure.core.exceptions.HttpResponseError: 415 """ 416 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 417 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGateway"] 418 lro_delay = kwargs.pop( 419 'polling_interval', 420 self._config.polling_interval 421 ) 422 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 423 if cont_token is None: 424 raw_result = await self._update_tags_initial( 425 resource_group_name=resource_group_name, 426 application_gateway_name=application_gateway_name, 427 parameters=parameters, 428 cls=lambda x,y,z: x, 429 **kwargs 430 ) 431 432 kwargs.pop('error_map', None) 433 kwargs.pop('content_type', None) 434 435 def get_long_running_output(pipeline_response): 436 deserialized = self._deserialize('ApplicationGateway', pipeline_response) 437 438 if cls: 439 return cls(pipeline_response, deserialized, {}) 440 return deserialized 441 442 path_format_arguments = { 443 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 444 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 445 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 446 } 447 448 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 449 elif polling is False: polling_method = AsyncNoPolling() 450 else: polling_method = polling 451 if cont_token: 452 return AsyncLROPoller.from_continuation_token( 453 polling_method=polling_method, 454 continuation_token=cont_token, 455 client=self._client, 456 deserialization_callback=get_long_running_output 457 ) 458 else: 459 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 460 begin_update_tags.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}'} # type: ignore 461 462 def list( 463 self, 464 resource_group_name: str, 465 **kwargs 466 ) -> AsyncIterable["_models.ApplicationGatewayListResult"]: 467 """Lists all application gateways in a resource group. 468 469 :param resource_group_name: The name of the resource group. 470 :type resource_group_name: str 471 :keyword callable cls: A custom type or function that will be passed the direct response 472 :return: An iterator like instance of either ApplicationGatewayListResult or the result of cls(response) 473 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2017_10_01.models.ApplicationGatewayListResult] 474 :raises: ~azure.core.exceptions.HttpResponseError 475 """ 476 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGatewayListResult"] 477 error_map = { 478 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 479 } 480 error_map.update(kwargs.pop('error_map', {})) 481 api_version = "2017-10-01" 482 accept = "application/json" 483 484 def prepare_request(next_link=None): 485 # Construct headers 486 header_parameters = {} # type: Dict[str, Any] 487 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 488 489 if not next_link: 490 # Construct URL 491 url = self.list.metadata['url'] # type: ignore 492 path_format_arguments = { 493 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 494 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 495 } 496 url = self._client.format_url(url, **path_format_arguments) 497 # Construct parameters 498 query_parameters = {} # type: Dict[str, Any] 499 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 500 501 request = self._client.get(url, query_parameters, header_parameters) 502 else: 503 url = next_link 504 query_parameters = {} # type: Dict[str, Any] 505 request = self._client.get(url, query_parameters, header_parameters) 506 return request 507 508 async def extract_data(pipeline_response): 509 deserialized = self._deserialize('ApplicationGatewayListResult', pipeline_response) 510 list_of_elem = deserialized.value 511 if cls: 512 list_of_elem = cls(list_of_elem) 513 return deserialized.next_link or None, AsyncList(list_of_elem) 514 515 async def get_next(next_link=None): 516 request = prepare_request(next_link) 517 518 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 519 response = pipeline_response.http_response 520 521 if response.status_code not in [200]: 522 map_error(status_code=response.status_code, response=response, error_map=error_map) 523 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 524 525 return pipeline_response 526 527 return AsyncItemPaged( 528 get_next, extract_data 529 ) 530 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways'} # type: ignore 531 532 def list_all( 533 self, 534 **kwargs 535 ) -> AsyncIterable["_models.ApplicationGatewayListResult"]: 536 """Gets all the application gateways in a subscription. 537 538 :keyword callable cls: A custom type or function that will be passed the direct response 539 :return: An iterator like instance of either ApplicationGatewayListResult or the result of cls(response) 540 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2017_10_01.models.ApplicationGatewayListResult] 541 :raises: ~azure.core.exceptions.HttpResponseError 542 """ 543 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGatewayListResult"] 544 error_map = { 545 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 546 } 547 error_map.update(kwargs.pop('error_map', {})) 548 api_version = "2017-10-01" 549 accept = "application/json" 550 551 def prepare_request(next_link=None): 552 # Construct headers 553 header_parameters = {} # type: Dict[str, Any] 554 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 555 556 if not next_link: 557 # Construct URL 558 url = self.list_all.metadata['url'] # type: ignore 559 path_format_arguments = { 560 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 561 } 562 url = self._client.format_url(url, **path_format_arguments) 563 # Construct parameters 564 query_parameters = {} # type: Dict[str, Any] 565 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 566 567 request = self._client.get(url, query_parameters, header_parameters) 568 else: 569 url = next_link 570 query_parameters = {} # type: Dict[str, Any] 571 request = self._client.get(url, query_parameters, header_parameters) 572 return request 573 574 async def extract_data(pipeline_response): 575 deserialized = self._deserialize('ApplicationGatewayListResult', pipeline_response) 576 list_of_elem = deserialized.value 577 if cls: 578 list_of_elem = cls(list_of_elem) 579 return deserialized.next_link or None, AsyncList(list_of_elem) 580 581 async def get_next(next_link=None): 582 request = prepare_request(next_link) 583 584 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 585 response = pipeline_response.http_response 586 587 if response.status_code not in [200]: 588 map_error(status_code=response.status_code, response=response, error_map=error_map) 589 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 590 591 return pipeline_response 592 593 return AsyncItemPaged( 594 get_next, extract_data 595 ) 596 list_all.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/applicationGateways'} # type: ignore 597 598 async def _start_initial( 599 self, 600 resource_group_name: str, 601 application_gateway_name: str, 602 **kwargs 603 ) -> None: 604 cls = kwargs.pop('cls', None) # type: ClsType[None] 605 error_map = { 606 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 607 } 608 error_map.update(kwargs.pop('error_map', {})) 609 api_version = "2017-10-01" 610 611 # Construct URL 612 url = self._start_initial.metadata['url'] # type: ignore 613 path_format_arguments = { 614 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 615 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 616 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 617 } 618 url = self._client.format_url(url, **path_format_arguments) 619 620 # Construct parameters 621 query_parameters = {} # type: Dict[str, Any] 622 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 623 624 # Construct headers 625 header_parameters = {} # type: Dict[str, Any] 626 627 request = self._client.post(url, query_parameters, header_parameters) 628 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 629 response = pipeline_response.http_response 630 631 if response.status_code not in [200, 202]: 632 map_error(status_code=response.status_code, response=response, error_map=error_map) 633 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 634 635 if cls: 636 return cls(pipeline_response, None, {}) 637 638 _start_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}/start'} # type: ignore 639 640 async def begin_start( 641 self, 642 resource_group_name: str, 643 application_gateway_name: str, 644 **kwargs 645 ) -> AsyncLROPoller[None]: 646 """Starts the specified application gateway. 647 648 :param resource_group_name: The name of the resource group. 649 :type resource_group_name: str 650 :param application_gateway_name: The name of the application gateway. 651 :type application_gateway_name: str 652 :keyword callable cls: A custom type or function that will be passed the direct response 653 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 654 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 655 False for no polling, or your own initialized polling object for a personal polling strategy. 656 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 657 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 658 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 659 :rtype: ~azure.core.polling.AsyncLROPoller[None] 660 :raises ~azure.core.exceptions.HttpResponseError: 661 """ 662 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 663 cls = kwargs.pop('cls', None) # type: ClsType[None] 664 lro_delay = kwargs.pop( 665 'polling_interval', 666 self._config.polling_interval 667 ) 668 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 669 if cont_token is None: 670 raw_result = await self._start_initial( 671 resource_group_name=resource_group_name, 672 application_gateway_name=application_gateway_name, 673 cls=lambda x,y,z: x, 674 **kwargs 675 ) 676 677 kwargs.pop('error_map', None) 678 kwargs.pop('content_type', None) 679 680 def get_long_running_output(pipeline_response): 681 if cls: 682 return cls(pipeline_response, None, {}) 683 684 path_format_arguments = { 685 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 686 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 687 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 688 } 689 690 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 691 elif polling is False: polling_method = AsyncNoPolling() 692 else: polling_method = polling 693 if cont_token: 694 return AsyncLROPoller.from_continuation_token( 695 polling_method=polling_method, 696 continuation_token=cont_token, 697 client=self._client, 698 deserialization_callback=get_long_running_output 699 ) 700 else: 701 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 702 begin_start.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}/start'} # type: ignore 703 704 async def _stop_initial( 705 self, 706 resource_group_name: str, 707 application_gateway_name: str, 708 **kwargs 709 ) -> None: 710 cls = kwargs.pop('cls', None) # type: ClsType[None] 711 error_map = { 712 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 713 } 714 error_map.update(kwargs.pop('error_map', {})) 715 api_version = "2017-10-01" 716 717 # Construct URL 718 url = self._stop_initial.metadata['url'] # type: ignore 719 path_format_arguments = { 720 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 721 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 722 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 723 } 724 url = self._client.format_url(url, **path_format_arguments) 725 726 # Construct parameters 727 query_parameters = {} # type: Dict[str, Any] 728 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 729 730 # Construct headers 731 header_parameters = {} # type: Dict[str, Any] 732 733 request = self._client.post(url, query_parameters, header_parameters) 734 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 735 response = pipeline_response.http_response 736 737 if response.status_code not in [200, 202]: 738 map_error(status_code=response.status_code, response=response, error_map=error_map) 739 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 740 741 if cls: 742 return cls(pipeline_response, None, {}) 743 744 _stop_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}/stop'} # type: ignore 745 746 async def begin_stop( 747 self, 748 resource_group_name: str, 749 application_gateway_name: str, 750 **kwargs 751 ) -> AsyncLROPoller[None]: 752 """Stops the specified application gateway in a resource group. 753 754 :param resource_group_name: The name of the resource group. 755 :type resource_group_name: str 756 :param application_gateway_name: The name of the application gateway. 757 :type application_gateway_name: str 758 :keyword callable cls: A custom type or function that will be passed the direct response 759 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 760 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 761 False for no polling, or your own initialized polling object for a personal polling strategy. 762 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 763 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 764 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 765 :rtype: ~azure.core.polling.AsyncLROPoller[None] 766 :raises ~azure.core.exceptions.HttpResponseError: 767 """ 768 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 769 cls = kwargs.pop('cls', None) # type: ClsType[None] 770 lro_delay = kwargs.pop( 771 'polling_interval', 772 self._config.polling_interval 773 ) 774 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 775 if cont_token is None: 776 raw_result = await self._stop_initial( 777 resource_group_name=resource_group_name, 778 application_gateway_name=application_gateway_name, 779 cls=lambda x,y,z: x, 780 **kwargs 781 ) 782 783 kwargs.pop('error_map', None) 784 kwargs.pop('content_type', None) 785 786 def get_long_running_output(pipeline_response): 787 if cls: 788 return cls(pipeline_response, None, {}) 789 790 path_format_arguments = { 791 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 792 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 793 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 794 } 795 796 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 797 elif polling is False: polling_method = AsyncNoPolling() 798 else: polling_method = polling 799 if cont_token: 800 return AsyncLROPoller.from_continuation_token( 801 polling_method=polling_method, 802 continuation_token=cont_token, 803 client=self._client, 804 deserialization_callback=get_long_running_output 805 ) 806 else: 807 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 808 begin_stop.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}/stop'} # type: ignore 809 810 async def _backend_health_initial( 811 self, 812 resource_group_name: str, 813 application_gateway_name: str, 814 expand: Optional[str] = None, 815 **kwargs 816 ) -> Optional["_models.ApplicationGatewayBackendHealth"]: 817 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.ApplicationGatewayBackendHealth"]] 818 error_map = { 819 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 820 } 821 error_map.update(kwargs.pop('error_map', {})) 822 api_version = "2017-10-01" 823 accept = "application/json" 824 825 # Construct URL 826 url = self._backend_health_initial.metadata['url'] # type: ignore 827 path_format_arguments = { 828 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 829 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 830 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 831 } 832 url = self._client.format_url(url, **path_format_arguments) 833 834 # Construct parameters 835 query_parameters = {} # type: Dict[str, Any] 836 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 837 if expand is not None: 838 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 839 840 # Construct headers 841 header_parameters = {} # type: Dict[str, Any] 842 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 843 844 request = self._client.post(url, query_parameters, header_parameters) 845 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 846 response = pipeline_response.http_response 847 848 if response.status_code not in [200, 202]: 849 map_error(status_code=response.status_code, response=response, error_map=error_map) 850 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 851 852 deserialized = None 853 if response.status_code == 200: 854 deserialized = self._deserialize('ApplicationGatewayBackendHealth', pipeline_response) 855 856 if cls: 857 return cls(pipeline_response, deserialized, {}) 858 859 return deserialized 860 _backend_health_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}/backendhealth'} # type: ignore 861 862 async def begin_backend_health( 863 self, 864 resource_group_name: str, 865 application_gateway_name: str, 866 expand: Optional[str] = None, 867 **kwargs 868 ) -> AsyncLROPoller["_models.ApplicationGatewayBackendHealth"]: 869 """Gets the backend health of the specified application gateway in a resource group. 870 871 :param resource_group_name: The name of the resource group. 872 :type resource_group_name: str 873 :param application_gateway_name: The name of the application gateway. 874 :type application_gateway_name: str 875 :param expand: Expands BackendAddressPool and BackendHttpSettings referenced in backend health. 876 :type expand: str 877 :keyword callable cls: A custom type or function that will be passed the direct response 878 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 879 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 880 False for no polling, or your own initialized polling object for a personal polling strategy. 881 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 882 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 883 :return: An instance of AsyncLROPoller that returns either ApplicationGatewayBackendHealth or the result of cls(response) 884 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2017_10_01.models.ApplicationGatewayBackendHealth] 885 :raises ~azure.core.exceptions.HttpResponseError: 886 """ 887 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 888 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGatewayBackendHealth"] 889 lro_delay = kwargs.pop( 890 'polling_interval', 891 self._config.polling_interval 892 ) 893 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 894 if cont_token is None: 895 raw_result = await self._backend_health_initial( 896 resource_group_name=resource_group_name, 897 application_gateway_name=application_gateway_name, 898 expand=expand, 899 cls=lambda x,y,z: x, 900 **kwargs 901 ) 902 903 kwargs.pop('error_map', None) 904 kwargs.pop('content_type', None) 905 906 def get_long_running_output(pipeline_response): 907 deserialized = self._deserialize('ApplicationGatewayBackendHealth', pipeline_response) 908 909 if cls: 910 return cls(pipeline_response, deserialized, {}) 911 return deserialized 912 913 path_format_arguments = { 914 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 915 'applicationGatewayName': self._serialize.url("application_gateway_name", application_gateway_name, 'str'), 916 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 917 } 918 919 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 920 elif polling is False: polling_method = AsyncNoPolling() 921 else: polling_method = polling 922 if cont_token: 923 return AsyncLROPoller.from_continuation_token( 924 polling_method=polling_method, 925 continuation_token=cont_token, 926 client=self._client, 927 deserialization_callback=get_long_running_output 928 ) 929 else: 930 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 931 begin_backend_health.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationGateways/{applicationGatewayName}/backendhealth'} # type: ignore 932 933 async def list_available_waf_rule_sets( 934 self, 935 **kwargs 936 ) -> "_models.ApplicationGatewayAvailableWafRuleSetsResult": 937 """Lists all available web application firewall rule sets. 938 939 :keyword callable cls: A custom type or function that will be passed the direct response 940 :return: ApplicationGatewayAvailableWafRuleSetsResult, or the result of cls(response) 941 :rtype: ~azure.mgmt.network.v2017_10_01.models.ApplicationGatewayAvailableWafRuleSetsResult 942 :raises: ~azure.core.exceptions.HttpResponseError 943 """ 944 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGatewayAvailableWafRuleSetsResult"] 945 error_map = { 946 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 947 } 948 error_map.update(kwargs.pop('error_map', {})) 949 api_version = "2017-10-01" 950 accept = "application/json" 951 952 # Construct URL 953 url = self.list_available_waf_rule_sets.metadata['url'] # type: ignore 954 path_format_arguments = { 955 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 956 } 957 url = self._client.format_url(url, **path_format_arguments) 958 959 # Construct parameters 960 query_parameters = {} # type: Dict[str, Any] 961 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 962 963 # Construct headers 964 header_parameters = {} # type: Dict[str, Any] 965 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 966 967 request = self._client.get(url, query_parameters, header_parameters) 968 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 969 response = pipeline_response.http_response 970 971 if response.status_code not in [200]: 972 map_error(status_code=response.status_code, response=response, error_map=error_map) 973 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 974 975 deserialized = self._deserialize('ApplicationGatewayAvailableWafRuleSetsResult', pipeline_response) 976 977 if cls: 978 return cls(pipeline_response, deserialized, {}) 979 980 return deserialized 981 list_available_waf_rule_sets.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/applicationGatewayAvailableWafRuleSets'} # type: ignore 982 983 async def list_available_ssl_options( 984 self, 985 **kwargs 986 ) -> "_models.ApplicationGatewayAvailableSslOptions": 987 """Lists available Ssl options for configuring Ssl policy. 988 989 :keyword callable cls: A custom type or function that will be passed the direct response 990 :return: ApplicationGatewayAvailableSslOptions, or the result of cls(response) 991 :rtype: ~azure.mgmt.network.v2017_10_01.models.ApplicationGatewayAvailableSslOptions 992 :raises: ~azure.core.exceptions.HttpResponseError 993 """ 994 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGatewayAvailableSslOptions"] 995 error_map = { 996 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 997 } 998 error_map.update(kwargs.pop('error_map', {})) 999 api_version = "2017-10-01" 1000 accept = "application/json" 1001 1002 # Construct URL 1003 url = self.list_available_ssl_options.metadata['url'] # type: ignore 1004 path_format_arguments = { 1005 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 1006 } 1007 url = self._client.format_url(url, **path_format_arguments) 1008 1009 # Construct parameters 1010 query_parameters = {} # type: Dict[str, Any] 1011 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1012 1013 # Construct headers 1014 header_parameters = {} # type: Dict[str, Any] 1015 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1016 1017 request = self._client.get(url, query_parameters, header_parameters) 1018 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 1019 response = pipeline_response.http_response 1020 1021 if response.status_code not in [200]: 1022 map_error(status_code=response.status_code, response=response, error_map=error_map) 1023 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1024 1025 deserialized = self._deserialize('ApplicationGatewayAvailableSslOptions', pipeline_response) 1026 1027 if cls: 1028 return cls(pipeline_response, deserialized, {}) 1029 1030 return deserialized 1031 list_available_ssl_options.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/applicationGatewayAvailableSslOptions/default'} # type: ignore 1032 1033 def list_available_ssl_predefined_policies( 1034 self, 1035 **kwargs 1036 ) -> AsyncIterable["_models.ApplicationGatewayAvailableSslPredefinedPolicies"]: 1037 """Lists all SSL predefined policies for configuring Ssl policy. 1038 1039 :keyword callable cls: A custom type or function that will be passed the direct response 1040 :return: An iterator like instance of either ApplicationGatewayAvailableSslPredefinedPolicies or the result of cls(response) 1041 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2017_10_01.models.ApplicationGatewayAvailableSslPredefinedPolicies] 1042 :raises: ~azure.core.exceptions.HttpResponseError 1043 """ 1044 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGatewayAvailableSslPredefinedPolicies"] 1045 error_map = { 1046 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1047 } 1048 error_map.update(kwargs.pop('error_map', {})) 1049 api_version = "2017-10-01" 1050 accept = "application/json" 1051 1052 def prepare_request(next_link=None): 1053 # Construct headers 1054 header_parameters = {} # type: Dict[str, Any] 1055 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1056 1057 if not next_link: 1058 # Construct URL 1059 url = self.list_available_ssl_predefined_policies.metadata['url'] # type: ignore 1060 path_format_arguments = { 1061 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 1062 } 1063 url = self._client.format_url(url, **path_format_arguments) 1064 # Construct parameters 1065 query_parameters = {} # type: Dict[str, Any] 1066 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1067 1068 request = self._client.get(url, query_parameters, header_parameters) 1069 else: 1070 url = next_link 1071 query_parameters = {} # type: Dict[str, Any] 1072 request = self._client.get(url, query_parameters, header_parameters) 1073 return request 1074 1075 async def extract_data(pipeline_response): 1076 deserialized = self._deserialize('ApplicationGatewayAvailableSslPredefinedPolicies', pipeline_response) 1077 list_of_elem = deserialized.value 1078 if cls: 1079 list_of_elem = cls(list_of_elem) 1080 return deserialized.next_link or None, AsyncList(list_of_elem) 1081 1082 async def get_next(next_link=None): 1083 request = prepare_request(next_link) 1084 1085 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 1086 response = pipeline_response.http_response 1087 1088 if response.status_code not in [200]: 1089 map_error(status_code=response.status_code, response=response, error_map=error_map) 1090 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1091 1092 return pipeline_response 1093 1094 return AsyncItemPaged( 1095 get_next, extract_data 1096 ) 1097 list_available_ssl_predefined_policies.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/applicationGatewayAvailableSslOptions/default/predefinedPolicies'} # type: ignore 1098 1099 async def get_ssl_predefined_policy( 1100 self, 1101 predefined_policy_name: str, 1102 **kwargs 1103 ) -> "_models.ApplicationGatewaySslPredefinedPolicy": 1104 """Gets Ssl predefined policy with the specified policy name. 1105 1106 :param predefined_policy_name: Name of Ssl predefined policy. 1107 :type predefined_policy_name: str 1108 :keyword callable cls: A custom type or function that will be passed the direct response 1109 :return: ApplicationGatewaySslPredefinedPolicy, or the result of cls(response) 1110 :rtype: ~azure.mgmt.network.v2017_10_01.models.ApplicationGatewaySslPredefinedPolicy 1111 :raises: ~azure.core.exceptions.HttpResponseError 1112 """ 1113 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationGatewaySslPredefinedPolicy"] 1114 error_map = { 1115 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1116 } 1117 error_map.update(kwargs.pop('error_map', {})) 1118 api_version = "2017-10-01" 1119 accept = "application/json" 1120 1121 # Construct URL 1122 url = self.get_ssl_predefined_policy.metadata['url'] # type: ignore 1123 path_format_arguments = { 1124 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 1125 'predefinedPolicyName': self._serialize.url("predefined_policy_name", predefined_policy_name, 'str'), 1126 } 1127 url = self._client.format_url(url, **path_format_arguments) 1128 1129 # Construct parameters 1130 query_parameters = {} # type: Dict[str, Any] 1131 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1132 1133 # Construct headers 1134 header_parameters = {} # type: Dict[str, Any] 1135 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1136 1137 request = self._client.get(url, query_parameters, header_parameters) 1138 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 1139 response = pipeline_response.http_response 1140 1141 if response.status_code not in [200]: 1142 map_error(status_code=response.status_code, response=response, error_map=error_map) 1143 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1144 1145 deserialized = self._deserialize('ApplicationGatewaySslPredefinedPolicy', pipeline_response) 1146 1147 if cls: 1148 return cls(pipeline_response, deserialized, {}) 1149 1150 return deserialized 1151 get_ssl_predefined_policy.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/applicationGatewayAvailableSslOptions/default/predefinedPolicies/{predefinedPolicyName}'} # type: ignore 1152