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