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