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