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