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, Optional, TypeVar, Union 24 25 T = TypeVar('T') 26 ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]] 27 28class VpnGatewaysOperations(object): 29 """VpnGatewaysOperations 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 get( 51 self, 52 resource_group_name, # type: str 53 gateway_name, # type: str 54 **kwargs # type: Any 55 ): 56 # type: (...) -> "_models.VpnGateway" 57 """Retrieves the details of a virtual wan vpn gateway. 58 59 :param resource_group_name: The resource group name of the VpnGateway. 60 :type resource_group_name: str 61 :param gateway_name: The name of the gateway. 62 :type gateway_name: str 63 :keyword callable cls: A custom type or function that will be passed the direct response 64 :return: VpnGateway, or the result of cls(response) 65 :rtype: ~azure.mgmt.network.v2019_11_01.models.VpnGateway 66 :raises: ~azure.core.exceptions.HttpResponseError 67 """ 68 cls = kwargs.pop('cls', None) # type: ClsType["_models.VpnGateway"] 69 error_map = { 70 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 71 } 72 error_map.update(kwargs.pop('error_map', {})) 73 api_version = "2019-11-01" 74 accept = "application/json" 75 76 # Construct URL 77 url = self.get.metadata['url'] # type: ignore 78 path_format_arguments = { 79 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 80 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 81 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 82 } 83 url = self._client.format_url(url, **path_format_arguments) 84 85 # Construct parameters 86 query_parameters = {} # type: Dict[str, Any] 87 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 88 89 # Construct headers 90 header_parameters = {} # type: Dict[str, Any] 91 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 92 93 request = self._client.get(url, query_parameters, header_parameters) 94 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 95 response = pipeline_response.http_response 96 97 if response.status_code not in [200]: 98 map_error(status_code=response.status_code, response=response, error_map=error_map) 99 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 100 101 deserialized = self._deserialize('VpnGateway', pipeline_response) 102 103 if cls: 104 return cls(pipeline_response, deserialized, {}) 105 106 return deserialized 107 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}'} # type: ignore 108 109 def _create_or_update_initial( 110 self, 111 resource_group_name, # type: str 112 gateway_name, # type: str 113 vpn_gateway_parameters, # type: "_models.VpnGateway" 114 **kwargs # type: Any 115 ): 116 # type: (...) -> "_models.VpnGateway" 117 cls = kwargs.pop('cls', None) # type: ClsType["_models.VpnGateway"] 118 error_map = { 119 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 120 } 121 error_map.update(kwargs.pop('error_map', {})) 122 api_version = "2019-11-01" 123 content_type = kwargs.pop("content_type", "application/json") 124 accept = "application/json" 125 126 # Construct URL 127 url = self._create_or_update_initial.metadata['url'] # type: ignore 128 path_format_arguments = { 129 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 130 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 131 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 132 } 133 url = self._client.format_url(url, **path_format_arguments) 134 135 # Construct parameters 136 query_parameters = {} # type: Dict[str, Any] 137 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 138 139 # Construct headers 140 header_parameters = {} # type: Dict[str, Any] 141 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 142 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 143 144 body_content_kwargs = {} # type: Dict[str, Any] 145 body_content = self._serialize.body(vpn_gateway_parameters, 'VpnGateway') 146 body_content_kwargs['content'] = body_content 147 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 148 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 149 response = pipeline_response.http_response 150 151 if response.status_code not in [200, 201]: 152 map_error(status_code=response.status_code, response=response, error_map=error_map) 153 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 154 155 if response.status_code == 200: 156 deserialized = self._deserialize('VpnGateway', pipeline_response) 157 158 if response.status_code == 201: 159 deserialized = self._deserialize('VpnGateway', pipeline_response) 160 161 if cls: 162 return cls(pipeline_response, deserialized, {}) 163 164 return deserialized 165 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}'} # type: ignore 166 167 def begin_create_or_update( 168 self, 169 resource_group_name, # type: str 170 gateway_name, # type: str 171 vpn_gateway_parameters, # type: "_models.VpnGateway" 172 **kwargs # type: Any 173 ): 174 # type: (...) -> LROPoller["_models.VpnGateway"] 175 """Creates a virtual wan vpn gateway if it doesn't exist else updates the existing gateway. 176 177 :param resource_group_name: The resource group name of the VpnGateway. 178 :type resource_group_name: str 179 :param gateway_name: The name of the gateway. 180 :type gateway_name: str 181 :param vpn_gateway_parameters: Parameters supplied to create or Update a virtual wan vpn 182 gateway. 183 :type vpn_gateway_parameters: ~azure.mgmt.network.v2019_11_01.models.VpnGateway 184 :keyword callable cls: A custom type or function that will be passed the direct response 185 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 186 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 187 False for no polling, or your own initialized polling object for a personal polling strategy. 188 :paramtype polling: bool or ~azure.core.polling.PollingMethod 189 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 190 :return: An instance of LROPoller that returns either VpnGateway or the result of cls(response) 191 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.network.v2019_11_01.models.VpnGateway] 192 :raises ~azure.core.exceptions.HttpResponseError: 193 """ 194 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 195 cls = kwargs.pop('cls', None) # type: ClsType["_models.VpnGateway"] 196 lro_delay = kwargs.pop( 197 'polling_interval', 198 self._config.polling_interval 199 ) 200 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 201 if cont_token is None: 202 raw_result = self._create_or_update_initial( 203 resource_group_name=resource_group_name, 204 gateway_name=gateway_name, 205 vpn_gateway_parameters=vpn_gateway_parameters, 206 cls=lambda x,y,z: x, 207 **kwargs 208 ) 209 210 kwargs.pop('error_map', None) 211 kwargs.pop('content_type', None) 212 213 def get_long_running_output(pipeline_response): 214 deserialized = self._deserialize('VpnGateway', pipeline_response) 215 216 if cls: 217 return cls(pipeline_response, deserialized, {}) 218 return deserialized 219 220 path_format_arguments = { 221 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 222 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 223 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 224 } 225 226 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 227 elif polling is False: polling_method = NoPolling() 228 else: polling_method = polling 229 if cont_token: 230 return LROPoller.from_continuation_token( 231 polling_method=polling_method, 232 continuation_token=cont_token, 233 client=self._client, 234 deserialization_callback=get_long_running_output 235 ) 236 else: 237 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 238 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}'} # type: ignore 239 240 def update_tags( 241 self, 242 resource_group_name, # type: str 243 gateway_name, # type: str 244 vpn_gateway_parameters, # type: "_models.TagsObject" 245 **kwargs # type: Any 246 ): 247 # type: (...) -> "_models.VpnGateway" 248 """Updates virtual wan vpn gateway tags. 249 250 :param resource_group_name: The resource group name of the VpnGateway. 251 :type resource_group_name: str 252 :param gateway_name: The name of the gateway. 253 :type gateway_name: str 254 :param vpn_gateway_parameters: Parameters supplied to update a virtual wan vpn gateway tags. 255 :type vpn_gateway_parameters: ~azure.mgmt.network.v2019_11_01.models.TagsObject 256 :keyword callable cls: A custom type or function that will be passed the direct response 257 :return: VpnGateway, or the result of cls(response) 258 :rtype: ~azure.mgmt.network.v2019_11_01.models.VpnGateway 259 :raises: ~azure.core.exceptions.HttpResponseError 260 """ 261 cls = kwargs.pop('cls', None) # type: ClsType["_models.VpnGateway"] 262 error_map = { 263 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 264 } 265 error_map.update(kwargs.pop('error_map', {})) 266 api_version = "2019-11-01" 267 content_type = kwargs.pop("content_type", "application/json") 268 accept = "application/json" 269 270 # Construct URL 271 url = self.update_tags.metadata['url'] # type: ignore 272 path_format_arguments = { 273 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 274 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 275 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 276 } 277 url = self._client.format_url(url, **path_format_arguments) 278 279 # Construct parameters 280 query_parameters = {} # type: Dict[str, Any] 281 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 282 283 # Construct headers 284 header_parameters = {} # type: Dict[str, Any] 285 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 286 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 287 288 body_content_kwargs = {} # type: Dict[str, Any] 289 body_content = self._serialize.body(vpn_gateway_parameters, 'TagsObject') 290 body_content_kwargs['content'] = body_content 291 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 292 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 293 response = pipeline_response.http_response 294 295 if response.status_code not in [200]: 296 map_error(status_code=response.status_code, response=response, error_map=error_map) 297 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 298 299 deserialized = self._deserialize('VpnGateway', pipeline_response) 300 301 if cls: 302 return cls(pipeline_response, deserialized, {}) 303 304 return deserialized 305 update_tags.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}'} # type: ignore 306 307 def _delete_initial( 308 self, 309 resource_group_name, # type: str 310 gateway_name, # type: str 311 **kwargs # type: Any 312 ): 313 # type: (...) -> None 314 cls = kwargs.pop('cls', None) # type: ClsType[None] 315 error_map = { 316 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 317 } 318 error_map.update(kwargs.pop('error_map', {})) 319 api_version = "2019-11-01" 320 accept = "application/json" 321 322 # Construct URL 323 url = self._delete_initial.metadata['url'] # type: ignore 324 path_format_arguments = { 325 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 326 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 327 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 328 } 329 url = self._client.format_url(url, **path_format_arguments) 330 331 # Construct parameters 332 query_parameters = {} # type: Dict[str, Any] 333 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 334 335 # Construct headers 336 header_parameters = {} # type: Dict[str, Any] 337 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 338 339 request = self._client.delete(url, query_parameters, header_parameters) 340 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 341 response = pipeline_response.http_response 342 343 if response.status_code not in [200, 202, 204]: 344 map_error(status_code=response.status_code, response=response, error_map=error_map) 345 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 346 347 if cls: 348 return cls(pipeline_response, None, {}) 349 350 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}'} # type: ignore 351 352 def begin_delete( 353 self, 354 resource_group_name, # type: str 355 gateway_name, # type: str 356 **kwargs # type: Any 357 ): 358 # type: (...) -> LROPoller[None] 359 """Deletes a virtual wan vpn gateway. 360 361 :param resource_group_name: The resource group name of the VpnGateway. 362 :type resource_group_name: str 363 :param gateway_name: The name of the gateway. 364 :type gateway_name: str 365 :keyword callable cls: A custom type or function that will be passed the direct response 366 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 367 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 368 False for no polling, or your own initialized polling object for a personal polling strategy. 369 :paramtype polling: bool or ~azure.core.polling.PollingMethod 370 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 371 :return: An instance of LROPoller that returns either None or the result of cls(response) 372 :rtype: ~azure.core.polling.LROPoller[None] 373 :raises ~azure.core.exceptions.HttpResponseError: 374 """ 375 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 376 cls = kwargs.pop('cls', None) # type: ClsType[None] 377 lro_delay = kwargs.pop( 378 'polling_interval', 379 self._config.polling_interval 380 ) 381 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 382 if cont_token is None: 383 raw_result = self._delete_initial( 384 resource_group_name=resource_group_name, 385 gateway_name=gateway_name, 386 cls=lambda x,y,z: x, 387 **kwargs 388 ) 389 390 kwargs.pop('error_map', None) 391 kwargs.pop('content_type', None) 392 393 def get_long_running_output(pipeline_response): 394 if cls: 395 return cls(pipeline_response, None, {}) 396 397 path_format_arguments = { 398 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 399 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 400 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 401 } 402 403 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 404 elif polling is False: polling_method = NoPolling() 405 else: polling_method = polling 406 if cont_token: 407 return LROPoller.from_continuation_token( 408 polling_method=polling_method, 409 continuation_token=cont_token, 410 client=self._client, 411 deserialization_callback=get_long_running_output 412 ) 413 else: 414 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 415 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}'} # type: ignore 416 417 def _reset_initial( 418 self, 419 resource_group_name, # type: str 420 gateway_name, # type: str 421 **kwargs # type: Any 422 ): 423 # type: (...) -> Optional["_models.VpnGateway"] 424 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.VpnGateway"]] 425 error_map = { 426 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 427 } 428 error_map.update(kwargs.pop('error_map', {})) 429 api_version = "2019-11-01" 430 accept = "application/json" 431 432 # Construct URL 433 url = self._reset_initial.metadata['url'] # type: ignore 434 path_format_arguments = { 435 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 436 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 437 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 438 } 439 url = self._client.format_url(url, **path_format_arguments) 440 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 # Construct headers 446 header_parameters = {} # type: Dict[str, Any] 447 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 448 449 request = self._client.post(url, query_parameters, header_parameters) 450 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 451 response = pipeline_response.http_response 452 453 if response.status_code not in [200, 202]: 454 map_error(status_code=response.status_code, response=response, error_map=error_map) 455 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 456 457 deserialized = None 458 if response.status_code == 200: 459 deserialized = self._deserialize('VpnGateway', pipeline_response) 460 461 if cls: 462 return cls(pipeline_response, deserialized, {}) 463 464 return deserialized 465 _reset_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}/reset'} # type: ignore 466 467 def begin_reset( 468 self, 469 resource_group_name, # type: str 470 gateway_name, # type: str 471 **kwargs # type: Any 472 ): 473 # type: (...) -> LROPoller["_models.VpnGateway"] 474 """Resets the primary of the vpn gateway in the specified resource group. 475 476 :param resource_group_name: The resource group name of the VpnGateway. 477 :type resource_group_name: str 478 :param gateway_name: The name of the gateway. 479 :type gateway_name: str 480 :keyword callable cls: A custom type or function that will be passed the direct response 481 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 482 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 483 False for no polling, or your own initialized polling object for a personal polling strategy. 484 :paramtype polling: bool or ~azure.core.polling.PollingMethod 485 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 486 :return: An instance of LROPoller that returns either VpnGateway or the result of cls(response) 487 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.network.v2019_11_01.models.VpnGateway] 488 :raises ~azure.core.exceptions.HttpResponseError: 489 """ 490 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 491 cls = kwargs.pop('cls', None) # type: ClsType["_models.VpnGateway"] 492 lro_delay = kwargs.pop( 493 'polling_interval', 494 self._config.polling_interval 495 ) 496 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 497 if cont_token is None: 498 raw_result = self._reset_initial( 499 resource_group_name=resource_group_name, 500 gateway_name=gateway_name, 501 cls=lambda x,y,z: x, 502 **kwargs 503 ) 504 505 kwargs.pop('error_map', None) 506 kwargs.pop('content_type', None) 507 508 def get_long_running_output(pipeline_response): 509 deserialized = self._deserialize('VpnGateway', pipeline_response) 510 511 if cls: 512 return cls(pipeline_response, deserialized, {}) 513 return deserialized 514 515 path_format_arguments = { 516 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 517 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 518 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 519 } 520 521 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 522 elif polling is False: polling_method = NoPolling() 523 else: polling_method = polling 524 if cont_token: 525 return LROPoller.from_continuation_token( 526 polling_method=polling_method, 527 continuation_token=cont_token, 528 client=self._client, 529 deserialization_callback=get_long_running_output 530 ) 531 else: 532 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 533 begin_reset.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}/reset'} # type: ignore 534 535 def list_by_resource_group( 536 self, 537 resource_group_name, # type: str 538 **kwargs # type: Any 539 ): 540 # type: (...) -> Iterable["_models.ListVpnGatewaysResult"] 541 """Lists all the VpnGateways in a resource group. 542 543 :param resource_group_name: The resource group name of the VpnGateway. 544 :type resource_group_name: str 545 :keyword callable cls: A custom type or function that will be passed the direct response 546 :return: An iterator like instance of either ListVpnGatewaysResult or the result of cls(response) 547 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_11_01.models.ListVpnGatewaysResult] 548 :raises: ~azure.core.exceptions.HttpResponseError 549 """ 550 cls = kwargs.pop('cls', None) # type: ClsType["_models.ListVpnGatewaysResult"] 551 error_map = { 552 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 553 } 554 error_map.update(kwargs.pop('error_map', {})) 555 api_version = "2019-11-01" 556 accept = "application/json" 557 558 def prepare_request(next_link=None): 559 # Construct headers 560 header_parameters = {} # type: Dict[str, Any] 561 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 562 563 if not next_link: 564 # Construct URL 565 url = self.list_by_resource_group.metadata['url'] # type: ignore 566 path_format_arguments = { 567 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 568 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 569 } 570 url = self._client.format_url(url, **path_format_arguments) 571 # Construct parameters 572 query_parameters = {} # type: Dict[str, Any] 573 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 574 575 request = self._client.get(url, query_parameters, header_parameters) 576 else: 577 url = next_link 578 query_parameters = {} # type: Dict[str, Any] 579 request = self._client.get(url, query_parameters, header_parameters) 580 return request 581 582 def extract_data(pipeline_response): 583 deserialized = self._deserialize('ListVpnGatewaysResult', pipeline_response) 584 list_of_elem = deserialized.value 585 if cls: 586 list_of_elem = cls(list_of_elem) 587 return deserialized.next_link or None, iter(list_of_elem) 588 589 def get_next(next_link=None): 590 request = prepare_request(next_link) 591 592 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 593 response = pipeline_response.http_response 594 595 if response.status_code not in [200]: 596 map_error(status_code=response.status_code, response=response, error_map=error_map) 597 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 598 599 return pipeline_response 600 601 return ItemPaged( 602 get_next, extract_data 603 ) 604 list_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways'} # type: ignore 605 606 def list( 607 self, 608 **kwargs # type: Any 609 ): 610 # type: (...) -> Iterable["_models.ListVpnGatewaysResult"] 611 """Lists all the VpnGateways in a subscription. 612 613 :keyword callable cls: A custom type or function that will be passed the direct response 614 :return: An iterator like instance of either ListVpnGatewaysResult or the result of cls(response) 615 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_11_01.models.ListVpnGatewaysResult] 616 :raises: ~azure.core.exceptions.HttpResponseError 617 """ 618 cls = kwargs.pop('cls', None) # type: ClsType["_models.ListVpnGatewaysResult"] 619 error_map = { 620 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 621 } 622 error_map.update(kwargs.pop('error_map', {})) 623 api_version = "2019-11-01" 624 accept = "application/json" 625 626 def prepare_request(next_link=None): 627 # Construct headers 628 header_parameters = {} # type: Dict[str, Any] 629 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 630 631 if not next_link: 632 # Construct URL 633 url = self.list.metadata['url'] # type: ignore 634 path_format_arguments = { 635 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 636 } 637 url = self._client.format_url(url, **path_format_arguments) 638 # Construct parameters 639 query_parameters = {} # type: Dict[str, Any] 640 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 641 642 request = self._client.get(url, query_parameters, header_parameters) 643 else: 644 url = next_link 645 query_parameters = {} # type: Dict[str, Any] 646 request = self._client.get(url, query_parameters, header_parameters) 647 return request 648 649 def extract_data(pipeline_response): 650 deserialized = self._deserialize('ListVpnGatewaysResult', pipeline_response) 651 list_of_elem = deserialized.value 652 if cls: 653 list_of_elem = cls(list_of_elem) 654 return deserialized.next_link or None, iter(list_of_elem) 655 656 def get_next(next_link=None): 657 request = prepare_request(next_link) 658 659 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 660 response = pipeline_response.http_response 661 662 if response.status_code not in [200]: 663 map_error(status_code=response.status_code, response=response, error_map=error_map) 664 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 665 666 return pipeline_response 667 668 return ItemPaged( 669 get_next, extract_data 670 ) 671 list.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/vpnGateways'} # type: ignore 672