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 VpnConnectionsOperations(object): 29 """VpnConnectionsOperations 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 connection_name, # type: str 55 **kwargs # type: Any 56 ): 57 # type: (...) -> "_models.VpnConnection" 58 """Retrieves the details of a vpn connection. 59 60 :param resource_group_name: The resource group name of the VpnGateway. 61 :type resource_group_name: str 62 :param gateway_name: The name of the gateway. 63 :type gateway_name: str 64 :param connection_name: The name of the vpn connection. 65 :type connection_name: str 66 :keyword callable cls: A custom type or function that will be passed the direct response 67 :return: VpnConnection, or the result of cls(response) 68 :rtype: ~azure.mgmt.network.v2019_11_01.models.VpnConnection 69 :raises: ~azure.core.exceptions.HttpResponseError 70 """ 71 cls = kwargs.pop('cls', None) # type: ClsType["_models.VpnConnection"] 72 error_map = { 73 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 74 } 75 error_map.update(kwargs.pop('error_map', {})) 76 api_version = "2019-11-01" 77 accept = "application/json" 78 79 # Construct URL 80 url = self.get.metadata['url'] # type: ignore 81 path_format_arguments = { 82 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 83 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 84 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 85 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 86 } 87 url = self._client.format_url(url, **path_format_arguments) 88 89 # Construct parameters 90 query_parameters = {} # type: Dict[str, Any] 91 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 92 93 # Construct headers 94 header_parameters = {} # type: Dict[str, Any] 95 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 96 97 request = self._client.get(url, query_parameters, header_parameters) 98 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 99 response = pipeline_response.http_response 100 101 if response.status_code not in [200]: 102 map_error(status_code=response.status_code, response=response, error_map=error_map) 103 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 104 105 deserialized = self._deserialize('VpnConnection', pipeline_response) 106 107 if cls: 108 return cls(pipeline_response, deserialized, {}) 109 110 return deserialized 111 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}/vpnConnections/{connectionName}'} # type: ignore 112 113 def _create_or_update_initial( 114 self, 115 resource_group_name, # type: str 116 gateway_name, # type: str 117 connection_name, # type: str 118 vpn_connection_parameters, # type: "_models.VpnConnection" 119 **kwargs # type: Any 120 ): 121 # type: (...) -> "_models.VpnConnection" 122 cls = kwargs.pop('cls', None) # type: ClsType["_models.VpnConnection"] 123 error_map = { 124 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 125 } 126 error_map.update(kwargs.pop('error_map', {})) 127 api_version = "2019-11-01" 128 content_type = kwargs.pop("content_type", "application/json") 129 accept = "application/json" 130 131 # Construct URL 132 url = self._create_or_update_initial.metadata['url'] # type: ignore 133 path_format_arguments = { 134 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 135 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 136 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 137 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 138 } 139 url = self._client.format_url(url, **path_format_arguments) 140 141 # Construct parameters 142 query_parameters = {} # type: Dict[str, Any] 143 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 144 145 # Construct headers 146 header_parameters = {} # type: Dict[str, Any] 147 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 148 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 149 150 body_content_kwargs = {} # type: Dict[str, Any] 151 body_content = self._serialize.body(vpn_connection_parameters, 'VpnConnection') 152 body_content_kwargs['content'] = body_content 153 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 154 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 155 response = pipeline_response.http_response 156 157 if response.status_code not in [200, 201]: 158 map_error(status_code=response.status_code, response=response, error_map=error_map) 159 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 160 161 if response.status_code == 200: 162 deserialized = self._deserialize('VpnConnection', pipeline_response) 163 164 if response.status_code == 201: 165 deserialized = self._deserialize('VpnConnection', pipeline_response) 166 167 if cls: 168 return cls(pipeline_response, deserialized, {}) 169 170 return deserialized 171 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}/vpnConnections/{connectionName}'} # type: ignore 172 173 def begin_create_or_update( 174 self, 175 resource_group_name, # type: str 176 gateway_name, # type: str 177 connection_name, # type: str 178 vpn_connection_parameters, # type: "_models.VpnConnection" 179 **kwargs # type: Any 180 ): 181 # type: (...) -> LROPoller["_models.VpnConnection"] 182 """Creates a vpn connection to a scalable vpn gateway if it doesn't exist else updates the 183 existing connection. 184 185 :param resource_group_name: The resource group name of the VpnGateway. 186 :type resource_group_name: str 187 :param gateway_name: The name of the gateway. 188 :type gateway_name: str 189 :param connection_name: The name of the connection. 190 :type connection_name: str 191 :param vpn_connection_parameters: Parameters supplied to create or Update a VPN Connection. 192 :type vpn_connection_parameters: ~azure.mgmt.network.v2019_11_01.models.VpnConnection 193 :keyword callable cls: A custom type or function that will be passed the direct response 194 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 195 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 196 False for no polling, or your own initialized polling object for a personal polling strategy. 197 :paramtype polling: bool or ~azure.core.polling.PollingMethod 198 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 199 :return: An instance of LROPoller that returns either VpnConnection or the result of cls(response) 200 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.network.v2019_11_01.models.VpnConnection] 201 :raises ~azure.core.exceptions.HttpResponseError: 202 """ 203 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 204 cls = kwargs.pop('cls', None) # type: ClsType["_models.VpnConnection"] 205 lro_delay = kwargs.pop( 206 'polling_interval', 207 self._config.polling_interval 208 ) 209 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 210 if cont_token is None: 211 raw_result = self._create_or_update_initial( 212 resource_group_name=resource_group_name, 213 gateway_name=gateway_name, 214 connection_name=connection_name, 215 vpn_connection_parameters=vpn_connection_parameters, 216 cls=lambda x,y,z: x, 217 **kwargs 218 ) 219 220 kwargs.pop('error_map', None) 221 kwargs.pop('content_type', None) 222 223 def get_long_running_output(pipeline_response): 224 deserialized = self._deserialize('VpnConnection', pipeline_response) 225 226 if cls: 227 return cls(pipeline_response, deserialized, {}) 228 return deserialized 229 230 path_format_arguments = { 231 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 232 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 233 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 234 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 235 } 236 237 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 238 elif polling is False: polling_method = NoPolling() 239 else: polling_method = polling 240 if cont_token: 241 return LROPoller.from_continuation_token( 242 polling_method=polling_method, 243 continuation_token=cont_token, 244 client=self._client, 245 deserialization_callback=get_long_running_output 246 ) 247 else: 248 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 249 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}/vpnConnections/{connectionName}'} # type: ignore 250 251 def _delete_initial( 252 self, 253 resource_group_name, # type: str 254 gateway_name, # type: str 255 connection_name, # type: str 256 **kwargs # type: Any 257 ): 258 # type: (...) -> None 259 cls = kwargs.pop('cls', None) # type: ClsType[None] 260 error_map = { 261 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 262 } 263 error_map.update(kwargs.pop('error_map', {})) 264 api_version = "2019-11-01" 265 accept = "application/json" 266 267 # Construct URL 268 url = self._delete_initial.metadata['url'] # type: ignore 269 path_format_arguments = { 270 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 271 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 272 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 273 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 274 } 275 url = self._client.format_url(url, **path_format_arguments) 276 277 # Construct parameters 278 query_parameters = {} # type: Dict[str, Any] 279 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 280 281 # Construct headers 282 header_parameters = {} # type: Dict[str, Any] 283 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 284 285 request = self._client.delete(url, query_parameters, header_parameters) 286 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 287 response = pipeline_response.http_response 288 289 if response.status_code not in [200, 202, 204]: 290 map_error(status_code=response.status_code, response=response, error_map=error_map) 291 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 292 293 if cls: 294 return cls(pipeline_response, None, {}) 295 296 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}/vpnConnections/{connectionName}'} # type: ignore 297 298 def begin_delete( 299 self, 300 resource_group_name, # type: str 301 gateway_name, # type: str 302 connection_name, # type: str 303 **kwargs # type: Any 304 ): 305 # type: (...) -> LROPoller[None] 306 """Deletes a vpn connection. 307 308 :param resource_group_name: The resource group name of the VpnGateway. 309 :type resource_group_name: str 310 :param gateway_name: The name of the gateway. 311 :type gateway_name: str 312 :param connection_name: The name of the connection. 313 :type connection_name: str 314 :keyword callable cls: A custom type or function that will be passed the direct response 315 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 316 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 317 False for no polling, or your own initialized polling object for a personal polling strategy. 318 :paramtype polling: bool or ~azure.core.polling.PollingMethod 319 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 320 :return: An instance of LROPoller that returns either None or the result of cls(response) 321 :rtype: ~azure.core.polling.LROPoller[None] 322 :raises ~azure.core.exceptions.HttpResponseError: 323 """ 324 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 325 cls = kwargs.pop('cls', None) # type: ClsType[None] 326 lro_delay = kwargs.pop( 327 'polling_interval', 328 self._config.polling_interval 329 ) 330 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 331 if cont_token is None: 332 raw_result = self._delete_initial( 333 resource_group_name=resource_group_name, 334 gateway_name=gateway_name, 335 connection_name=connection_name, 336 cls=lambda x,y,z: x, 337 **kwargs 338 ) 339 340 kwargs.pop('error_map', None) 341 kwargs.pop('content_type', None) 342 343 def get_long_running_output(pipeline_response): 344 if cls: 345 return cls(pipeline_response, None, {}) 346 347 path_format_arguments = { 348 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 349 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 350 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 351 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 352 } 353 354 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 355 elif polling is False: polling_method = NoPolling() 356 else: polling_method = polling 357 if cont_token: 358 return LROPoller.from_continuation_token( 359 polling_method=polling_method, 360 continuation_token=cont_token, 361 client=self._client, 362 deserialization_callback=get_long_running_output 363 ) 364 else: 365 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 366 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}/vpnConnections/{connectionName}'} # type: ignore 367 368 def list_by_vpn_gateway( 369 self, 370 resource_group_name, # type: str 371 gateway_name, # type: str 372 **kwargs # type: Any 373 ): 374 # type: (...) -> Iterable["_models.ListVpnConnectionsResult"] 375 """Retrieves all vpn connections for a particular virtual wan vpn gateway. 376 377 :param resource_group_name: The resource group name of the VpnGateway. 378 :type resource_group_name: str 379 :param gateway_name: The name of the gateway. 380 :type gateway_name: str 381 :keyword callable cls: A custom type or function that will be passed the direct response 382 :return: An iterator like instance of either ListVpnConnectionsResult or the result of cls(response) 383 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_11_01.models.ListVpnConnectionsResult] 384 :raises: ~azure.core.exceptions.HttpResponseError 385 """ 386 cls = kwargs.pop('cls', None) # type: ClsType["_models.ListVpnConnectionsResult"] 387 error_map = { 388 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 389 } 390 error_map.update(kwargs.pop('error_map', {})) 391 api_version = "2019-11-01" 392 accept = "application/json" 393 394 def prepare_request(next_link=None): 395 # Construct headers 396 header_parameters = {} # type: Dict[str, Any] 397 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 398 399 if not next_link: 400 # Construct URL 401 url = self.list_by_vpn_gateway.metadata['url'] # type: ignore 402 path_format_arguments = { 403 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 404 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 405 'gatewayName': self._serialize.url("gateway_name", gateway_name, 'str'), 406 } 407 url = self._client.format_url(url, **path_format_arguments) 408 # Construct parameters 409 query_parameters = {} # type: Dict[str, Any] 410 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 411 412 request = self._client.get(url, query_parameters, header_parameters) 413 else: 414 url = next_link 415 query_parameters = {} # type: Dict[str, Any] 416 request = self._client.get(url, query_parameters, header_parameters) 417 return request 418 419 def extract_data(pipeline_response): 420 deserialized = self._deserialize('ListVpnConnectionsResult', pipeline_response) 421 list_of_elem = deserialized.value 422 if cls: 423 list_of_elem = cls(list_of_elem) 424 return deserialized.next_link or None, iter(list_of_elem) 425 426 def get_next(next_link=None): 427 request = prepare_request(next_link) 428 429 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 430 response = pipeline_response.http_response 431 432 if response.status_code not in [200]: 433 map_error(status_code=response.status_code, response=response, error_map=error_map) 434 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 435 436 return pipeline_response 437 438 return ItemPaged( 439 get_next, extract_data 440 ) 441 list_by_vpn_gateway.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/vpnGateways/{gatewayName}/vpnConnections'} # type: ignore 442