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 Any, AsyncIterable, Callable, Dict, Generic, Optional, TypeVar, Union 9import warnings 10 11from azure.core.async_paging import AsyncItemPaged, AsyncList 12from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error 13from azure.core.pipeline import PipelineResponse 14from azure.core.pipeline.transport import AsyncHttpResponse, HttpRequest 15from azure.core.polling import AsyncLROPoller, AsyncNoPolling, AsyncPollingMethod 16from azure.mgmt.core.exceptions import ARMErrorFormat 17from azure.mgmt.core.polling.async_arm_polling import AsyncARMPolling 18 19from ... import models as _models 20 21T = TypeVar('T') 22ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]] 23 24class ExpressRouteCrossConnectionsOperations: 25 """ExpressRouteCrossConnectionsOperations async operations. 26 27 You should not instantiate this class directly. Instead, you should create a Client instance that 28 instantiates it for you and attaches it as an attribute. 29 30 :ivar models: Alias to model classes used in this operation group. 31 :type models: ~azure.mgmt.network.v2020_06_01.models 32 :param client: Client for service requests. 33 :param config: Configuration of service client. 34 :param serializer: An object model serializer. 35 :param deserializer: An object model deserializer. 36 """ 37 38 models = _models 39 40 def __init__(self, client, config, serializer, deserializer) -> None: 41 self._client = client 42 self._serialize = serializer 43 self._deserialize = deserializer 44 self._config = config 45 46 def list( 47 self, 48 **kwargs 49 ) -> AsyncIterable["_models.ExpressRouteCrossConnectionListResult"]: 50 """Retrieves all the ExpressRouteCrossConnections in a subscription. 51 52 :keyword callable cls: A custom type or function that will be passed the direct response 53 :return: An iterator like instance of either ExpressRouteCrossConnectionListResult or the result of cls(response) 54 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_06_01.models.ExpressRouteCrossConnectionListResult] 55 :raises: ~azure.core.exceptions.HttpResponseError 56 """ 57 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCrossConnectionListResult"] 58 error_map = { 59 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 60 } 61 error_map.update(kwargs.pop('error_map', {})) 62 api_version = "2020-06-01" 63 accept = "application/json" 64 65 def prepare_request(next_link=None): 66 # Construct headers 67 header_parameters = {} # type: Dict[str, Any] 68 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 69 70 if not next_link: 71 # Construct URL 72 url = self.list.metadata['url'] # type: ignore 73 path_format_arguments = { 74 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 75 } 76 url = self._client.format_url(url, **path_format_arguments) 77 # Construct parameters 78 query_parameters = {} # type: Dict[str, Any] 79 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 80 81 request = self._client.get(url, query_parameters, header_parameters) 82 else: 83 url = next_link 84 query_parameters = {} # type: Dict[str, Any] 85 request = self._client.get(url, query_parameters, header_parameters) 86 return request 87 88 async def extract_data(pipeline_response): 89 deserialized = self._deserialize('ExpressRouteCrossConnectionListResult', pipeline_response) 90 list_of_elem = deserialized.value 91 if cls: 92 list_of_elem = cls(list_of_elem) 93 return deserialized.next_link or None, AsyncList(list_of_elem) 94 95 async def get_next(next_link=None): 96 request = prepare_request(next_link) 97 98 pipeline_response = await 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 return pipeline_response 106 107 return AsyncItemPaged( 108 get_next, extract_data 109 ) 110 list.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/expressRouteCrossConnections'} # type: ignore 111 112 def list_by_resource_group( 113 self, 114 resource_group_name: str, 115 **kwargs 116 ) -> AsyncIterable["_models.ExpressRouteCrossConnectionListResult"]: 117 """Retrieves all the ExpressRouteCrossConnections in a resource group. 118 119 :param resource_group_name: The name of the resource group. 120 :type resource_group_name: str 121 :keyword callable cls: A custom type or function that will be passed the direct response 122 :return: An iterator like instance of either ExpressRouteCrossConnectionListResult or the result of cls(response) 123 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_06_01.models.ExpressRouteCrossConnectionListResult] 124 :raises: ~azure.core.exceptions.HttpResponseError 125 """ 126 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCrossConnectionListResult"] 127 error_map = { 128 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 129 } 130 error_map.update(kwargs.pop('error_map', {})) 131 api_version = "2020-06-01" 132 accept = "application/json" 133 134 def prepare_request(next_link=None): 135 # Construct headers 136 header_parameters = {} # type: Dict[str, Any] 137 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 138 139 if not next_link: 140 # Construct URL 141 url = self.list_by_resource_group.metadata['url'] # type: ignore 142 path_format_arguments = { 143 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 144 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 145 } 146 url = self._client.format_url(url, **path_format_arguments) 147 # Construct parameters 148 query_parameters = {} # type: Dict[str, Any] 149 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 150 151 request = self._client.get(url, query_parameters, header_parameters) 152 else: 153 url = next_link 154 query_parameters = {} # type: Dict[str, Any] 155 request = self._client.get(url, query_parameters, header_parameters) 156 return request 157 158 async def extract_data(pipeline_response): 159 deserialized = self._deserialize('ExpressRouteCrossConnectionListResult', pipeline_response) 160 list_of_elem = deserialized.value 161 if cls: 162 list_of_elem = cls(list_of_elem) 163 return deserialized.next_link or None, AsyncList(list_of_elem) 164 165 async def get_next(next_link=None): 166 request = prepare_request(next_link) 167 168 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 169 response = pipeline_response.http_response 170 171 if response.status_code not in [200]: 172 map_error(status_code=response.status_code, response=response, error_map=error_map) 173 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 174 175 return pipeline_response 176 177 return AsyncItemPaged( 178 get_next, extract_data 179 ) 180 list_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections'} # type: ignore 181 182 async def get( 183 self, 184 resource_group_name: str, 185 cross_connection_name: str, 186 **kwargs 187 ) -> "_models.ExpressRouteCrossConnection": 188 """Gets details about the specified ExpressRouteCrossConnection. 189 190 :param resource_group_name: The name of the resource group (peering location of the circuit). 191 :type resource_group_name: str 192 :param cross_connection_name: The name of the ExpressRouteCrossConnection (service key of the 193 circuit). 194 :type cross_connection_name: str 195 :keyword callable cls: A custom type or function that will be passed the direct response 196 :return: ExpressRouteCrossConnection, or the result of cls(response) 197 :rtype: ~azure.mgmt.network.v2020_06_01.models.ExpressRouteCrossConnection 198 :raises: ~azure.core.exceptions.HttpResponseError 199 """ 200 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCrossConnection"] 201 error_map = { 202 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 203 } 204 error_map.update(kwargs.pop('error_map', {})) 205 api_version = "2020-06-01" 206 accept = "application/json" 207 208 # Construct URL 209 url = self.get.metadata['url'] # type: ignore 210 path_format_arguments = { 211 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 212 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 213 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 214 } 215 url = self._client.format_url(url, **path_format_arguments) 216 217 # Construct parameters 218 query_parameters = {} # type: Dict[str, Any] 219 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 220 221 # Construct headers 222 header_parameters = {} # type: Dict[str, Any] 223 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 224 225 request = self._client.get(url, query_parameters, header_parameters) 226 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 227 response = pipeline_response.http_response 228 229 if response.status_code not in [200]: 230 map_error(status_code=response.status_code, response=response, error_map=error_map) 231 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 232 233 deserialized = self._deserialize('ExpressRouteCrossConnection', pipeline_response) 234 235 if cls: 236 return cls(pipeline_response, deserialized, {}) 237 238 return deserialized 239 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}'} # type: ignore 240 241 async def _create_or_update_initial( 242 self, 243 resource_group_name: str, 244 cross_connection_name: str, 245 parameters: "_models.ExpressRouteCrossConnection", 246 **kwargs 247 ) -> "_models.ExpressRouteCrossConnection": 248 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCrossConnection"] 249 error_map = { 250 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 251 } 252 error_map.update(kwargs.pop('error_map', {})) 253 api_version = "2020-06-01" 254 content_type = kwargs.pop("content_type", "application/json") 255 accept = "application/json" 256 257 # Construct URL 258 url = self._create_or_update_initial.metadata['url'] # type: ignore 259 path_format_arguments = { 260 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 261 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 262 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 263 } 264 url = self._client.format_url(url, **path_format_arguments) 265 266 # Construct parameters 267 query_parameters = {} # type: Dict[str, Any] 268 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 269 270 # Construct headers 271 header_parameters = {} # type: Dict[str, Any] 272 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 273 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 274 275 body_content_kwargs = {} # type: Dict[str, Any] 276 body_content = self._serialize.body(parameters, 'ExpressRouteCrossConnection') 277 body_content_kwargs['content'] = body_content 278 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 279 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 280 response = pipeline_response.http_response 281 282 if response.status_code not in [200]: 283 map_error(status_code=response.status_code, response=response, error_map=error_map) 284 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 285 286 deserialized = self._deserialize('ExpressRouteCrossConnection', pipeline_response) 287 288 if cls: 289 return cls(pipeline_response, deserialized, {}) 290 291 return deserialized 292 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}'} # type: ignore 293 294 async def begin_create_or_update( 295 self, 296 resource_group_name: str, 297 cross_connection_name: str, 298 parameters: "_models.ExpressRouteCrossConnection", 299 **kwargs 300 ) -> AsyncLROPoller["_models.ExpressRouteCrossConnection"]: 301 """Update the specified ExpressRouteCrossConnection. 302 303 :param resource_group_name: The name of the resource group. 304 :type resource_group_name: str 305 :param cross_connection_name: The name of the ExpressRouteCrossConnection. 306 :type cross_connection_name: str 307 :param parameters: Parameters supplied to the update express route crossConnection operation. 308 :type parameters: ~azure.mgmt.network.v2020_06_01.models.ExpressRouteCrossConnection 309 :keyword callable cls: A custom type or function that will be passed the direct response 310 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 311 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 312 False for no polling, or your own initialized polling object for a personal polling strategy. 313 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 314 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 315 :return: An instance of AsyncLROPoller that returns either ExpressRouteCrossConnection or the result of cls(response) 316 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_06_01.models.ExpressRouteCrossConnection] 317 :raises ~azure.core.exceptions.HttpResponseError: 318 """ 319 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 320 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCrossConnection"] 321 lro_delay = kwargs.pop( 322 'polling_interval', 323 self._config.polling_interval 324 ) 325 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 326 if cont_token is None: 327 raw_result = await self._create_or_update_initial( 328 resource_group_name=resource_group_name, 329 cross_connection_name=cross_connection_name, 330 parameters=parameters, 331 cls=lambda x,y,z: x, 332 **kwargs 333 ) 334 335 kwargs.pop('error_map', None) 336 kwargs.pop('content_type', None) 337 338 def get_long_running_output(pipeline_response): 339 deserialized = self._deserialize('ExpressRouteCrossConnection', pipeline_response) 340 341 if cls: 342 return cls(pipeline_response, deserialized, {}) 343 return deserialized 344 345 path_format_arguments = { 346 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 347 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 348 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 349 } 350 351 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 352 elif polling is False: polling_method = AsyncNoPolling() 353 else: polling_method = polling 354 if cont_token: 355 return AsyncLROPoller.from_continuation_token( 356 polling_method=polling_method, 357 continuation_token=cont_token, 358 client=self._client, 359 deserialization_callback=get_long_running_output 360 ) 361 else: 362 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 363 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}'} # type: ignore 364 365 async def update_tags( 366 self, 367 resource_group_name: str, 368 cross_connection_name: str, 369 cross_connection_parameters: "_models.TagsObject", 370 **kwargs 371 ) -> "_models.ExpressRouteCrossConnection": 372 """Updates an express route cross connection tags. 373 374 :param resource_group_name: The name of the resource group. 375 :type resource_group_name: str 376 :param cross_connection_name: The name of the cross connection. 377 :type cross_connection_name: str 378 :param cross_connection_parameters: Parameters supplied to update express route cross 379 connection tags. 380 :type cross_connection_parameters: ~azure.mgmt.network.v2020_06_01.models.TagsObject 381 :keyword callable cls: A custom type or function that will be passed the direct response 382 :return: ExpressRouteCrossConnection, or the result of cls(response) 383 :rtype: ~azure.mgmt.network.v2020_06_01.models.ExpressRouteCrossConnection 384 :raises: ~azure.core.exceptions.HttpResponseError 385 """ 386 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCrossConnection"] 387 error_map = { 388 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 389 } 390 error_map.update(kwargs.pop('error_map', {})) 391 api_version = "2020-06-01" 392 content_type = kwargs.pop("content_type", "application/json") 393 accept = "application/json" 394 395 # Construct URL 396 url = self.update_tags.metadata['url'] # type: ignore 397 path_format_arguments = { 398 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 399 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 400 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 401 } 402 url = self._client.format_url(url, **path_format_arguments) 403 404 # Construct parameters 405 query_parameters = {} # type: Dict[str, Any] 406 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 407 408 # Construct headers 409 header_parameters = {} # type: Dict[str, Any] 410 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 411 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 412 413 body_content_kwargs = {} # type: Dict[str, Any] 414 body_content = self._serialize.body(cross_connection_parameters, 'TagsObject') 415 body_content_kwargs['content'] = body_content 416 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 417 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 418 response = pipeline_response.http_response 419 420 if response.status_code not in [200]: 421 map_error(status_code=response.status_code, response=response, error_map=error_map) 422 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 423 424 deserialized = self._deserialize('ExpressRouteCrossConnection', pipeline_response) 425 426 if cls: 427 return cls(pipeline_response, deserialized, {}) 428 429 return deserialized 430 update_tags.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}'} # type: ignore 431 432 async def _list_arp_table_initial( 433 self, 434 resource_group_name: str, 435 cross_connection_name: str, 436 peering_name: str, 437 device_path: str, 438 **kwargs 439 ) -> Optional["_models.ExpressRouteCircuitsArpTableListResult"]: 440 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.ExpressRouteCircuitsArpTableListResult"]] 441 error_map = { 442 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 443 } 444 error_map.update(kwargs.pop('error_map', {})) 445 api_version = "2020-06-01" 446 accept = "application/json" 447 448 # Construct URL 449 url = self._list_arp_table_initial.metadata['url'] # type: ignore 450 path_format_arguments = { 451 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 452 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 453 'peeringName': self._serialize.url("peering_name", peering_name, 'str'), 454 'devicePath': self._serialize.url("device_path", device_path, 'str'), 455 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 456 } 457 url = self._client.format_url(url, **path_format_arguments) 458 459 # Construct parameters 460 query_parameters = {} # type: Dict[str, Any] 461 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 462 463 # Construct headers 464 header_parameters = {} # type: Dict[str, Any] 465 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 466 467 request = self._client.post(url, query_parameters, header_parameters) 468 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 469 response = pipeline_response.http_response 470 471 if response.status_code not in [200, 202]: 472 map_error(status_code=response.status_code, response=response, error_map=error_map) 473 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 474 475 deserialized = None 476 if response.status_code == 200: 477 deserialized = self._deserialize('ExpressRouteCircuitsArpTableListResult', pipeline_response) 478 479 if cls: 480 return cls(pipeline_response, deserialized, {}) 481 482 return deserialized 483 _list_arp_table_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}/peerings/{peeringName}/arpTables/{devicePath}'} # type: ignore 484 485 async def begin_list_arp_table( 486 self, 487 resource_group_name: str, 488 cross_connection_name: str, 489 peering_name: str, 490 device_path: str, 491 **kwargs 492 ) -> AsyncLROPoller["_models.ExpressRouteCircuitsArpTableListResult"]: 493 """Gets the currently advertised ARP table associated with the express route cross connection in a 494 resource group. 495 496 :param resource_group_name: The name of the resource group. 497 :type resource_group_name: str 498 :param cross_connection_name: The name of the ExpressRouteCrossConnection. 499 :type cross_connection_name: str 500 :param peering_name: The name of the peering. 501 :type peering_name: str 502 :param device_path: The path of the device. 503 :type device_path: str 504 :keyword callable cls: A custom type or function that will be passed the direct response 505 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 506 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 507 False for no polling, or your own initialized polling object for a personal polling strategy. 508 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 509 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 510 :return: An instance of AsyncLROPoller that returns either ExpressRouteCircuitsArpTableListResult or the result of cls(response) 511 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_06_01.models.ExpressRouteCircuitsArpTableListResult] 512 :raises ~azure.core.exceptions.HttpResponseError: 513 """ 514 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 515 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCircuitsArpTableListResult"] 516 lro_delay = kwargs.pop( 517 'polling_interval', 518 self._config.polling_interval 519 ) 520 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 521 if cont_token is None: 522 raw_result = await self._list_arp_table_initial( 523 resource_group_name=resource_group_name, 524 cross_connection_name=cross_connection_name, 525 peering_name=peering_name, 526 device_path=device_path, 527 cls=lambda x,y,z: x, 528 **kwargs 529 ) 530 531 kwargs.pop('error_map', None) 532 kwargs.pop('content_type', None) 533 534 def get_long_running_output(pipeline_response): 535 deserialized = self._deserialize('ExpressRouteCircuitsArpTableListResult', pipeline_response) 536 537 if cls: 538 return cls(pipeline_response, deserialized, {}) 539 return deserialized 540 541 path_format_arguments = { 542 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 543 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 544 'peeringName': self._serialize.url("peering_name", peering_name, 'str'), 545 'devicePath': self._serialize.url("device_path", device_path, 'str'), 546 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 547 } 548 549 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 550 elif polling is False: polling_method = AsyncNoPolling() 551 else: polling_method = polling 552 if cont_token: 553 return AsyncLROPoller.from_continuation_token( 554 polling_method=polling_method, 555 continuation_token=cont_token, 556 client=self._client, 557 deserialization_callback=get_long_running_output 558 ) 559 else: 560 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 561 begin_list_arp_table.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}/peerings/{peeringName}/arpTables/{devicePath}'} # type: ignore 562 563 async def _list_routes_table_summary_initial( 564 self, 565 resource_group_name: str, 566 cross_connection_name: str, 567 peering_name: str, 568 device_path: str, 569 **kwargs 570 ) -> Optional["_models.ExpressRouteCrossConnectionsRoutesTableSummaryListResult"]: 571 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.ExpressRouteCrossConnectionsRoutesTableSummaryListResult"]] 572 error_map = { 573 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 574 } 575 error_map.update(kwargs.pop('error_map', {})) 576 api_version = "2020-06-01" 577 accept = "application/json" 578 579 # Construct URL 580 url = self._list_routes_table_summary_initial.metadata['url'] # type: ignore 581 path_format_arguments = { 582 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 583 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 584 'peeringName': self._serialize.url("peering_name", peering_name, 'str'), 585 'devicePath': self._serialize.url("device_path", device_path, 'str'), 586 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 587 } 588 url = self._client.format_url(url, **path_format_arguments) 589 590 # Construct parameters 591 query_parameters = {} # type: Dict[str, Any] 592 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 593 594 # Construct headers 595 header_parameters = {} # type: Dict[str, Any] 596 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 597 598 request = self._client.post(url, query_parameters, header_parameters) 599 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 600 response = pipeline_response.http_response 601 602 if response.status_code not in [200, 202]: 603 map_error(status_code=response.status_code, response=response, error_map=error_map) 604 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 605 606 deserialized = None 607 if response.status_code == 200: 608 deserialized = self._deserialize('ExpressRouteCrossConnectionsRoutesTableSummaryListResult', pipeline_response) 609 610 if cls: 611 return cls(pipeline_response, deserialized, {}) 612 613 return deserialized 614 _list_routes_table_summary_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}/peerings/{peeringName}/routeTablesSummary/{devicePath}'} # type: ignore 615 616 async def begin_list_routes_table_summary( 617 self, 618 resource_group_name: str, 619 cross_connection_name: str, 620 peering_name: str, 621 device_path: str, 622 **kwargs 623 ) -> AsyncLROPoller["_models.ExpressRouteCrossConnectionsRoutesTableSummaryListResult"]: 624 """Gets the route table summary associated with the express route cross connection in a resource 625 group. 626 627 :param resource_group_name: The name of the resource group. 628 :type resource_group_name: str 629 :param cross_connection_name: The name of the ExpressRouteCrossConnection. 630 :type cross_connection_name: str 631 :param peering_name: The name of the peering. 632 :type peering_name: str 633 :param device_path: The path of the device. 634 :type device_path: str 635 :keyword callable cls: A custom type or function that will be passed the direct response 636 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 637 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 638 False for no polling, or your own initialized polling object for a personal polling strategy. 639 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 640 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 641 :return: An instance of AsyncLROPoller that returns either ExpressRouteCrossConnectionsRoutesTableSummaryListResult or the result of cls(response) 642 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_06_01.models.ExpressRouteCrossConnectionsRoutesTableSummaryListResult] 643 :raises ~azure.core.exceptions.HttpResponseError: 644 """ 645 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 646 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCrossConnectionsRoutesTableSummaryListResult"] 647 lro_delay = kwargs.pop( 648 'polling_interval', 649 self._config.polling_interval 650 ) 651 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 652 if cont_token is None: 653 raw_result = await self._list_routes_table_summary_initial( 654 resource_group_name=resource_group_name, 655 cross_connection_name=cross_connection_name, 656 peering_name=peering_name, 657 device_path=device_path, 658 cls=lambda x,y,z: x, 659 **kwargs 660 ) 661 662 kwargs.pop('error_map', None) 663 kwargs.pop('content_type', None) 664 665 def get_long_running_output(pipeline_response): 666 deserialized = self._deserialize('ExpressRouteCrossConnectionsRoutesTableSummaryListResult', pipeline_response) 667 668 if cls: 669 return cls(pipeline_response, deserialized, {}) 670 return deserialized 671 672 path_format_arguments = { 673 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 674 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 675 'peeringName': self._serialize.url("peering_name", peering_name, 'str'), 676 'devicePath': self._serialize.url("device_path", device_path, 'str'), 677 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 678 } 679 680 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 681 elif polling is False: polling_method = AsyncNoPolling() 682 else: polling_method = polling 683 if cont_token: 684 return AsyncLROPoller.from_continuation_token( 685 polling_method=polling_method, 686 continuation_token=cont_token, 687 client=self._client, 688 deserialization_callback=get_long_running_output 689 ) 690 else: 691 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 692 begin_list_routes_table_summary.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}/peerings/{peeringName}/routeTablesSummary/{devicePath}'} # type: ignore 693 694 async def _list_routes_table_initial( 695 self, 696 resource_group_name: str, 697 cross_connection_name: str, 698 peering_name: str, 699 device_path: str, 700 **kwargs 701 ) -> Optional["_models.ExpressRouteCircuitsRoutesTableListResult"]: 702 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.ExpressRouteCircuitsRoutesTableListResult"]] 703 error_map = { 704 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 705 } 706 error_map.update(kwargs.pop('error_map', {})) 707 api_version = "2020-06-01" 708 accept = "application/json" 709 710 # Construct URL 711 url = self._list_routes_table_initial.metadata['url'] # type: ignore 712 path_format_arguments = { 713 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 714 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 715 'peeringName': self._serialize.url("peering_name", peering_name, 'str'), 716 'devicePath': self._serialize.url("device_path", device_path, 'str'), 717 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 718 } 719 url = self._client.format_url(url, **path_format_arguments) 720 721 # Construct parameters 722 query_parameters = {} # type: Dict[str, Any] 723 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 724 725 # Construct headers 726 header_parameters = {} # type: Dict[str, Any] 727 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 728 729 request = self._client.post(url, query_parameters, header_parameters) 730 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 731 response = pipeline_response.http_response 732 733 if response.status_code not in [200, 202]: 734 map_error(status_code=response.status_code, response=response, error_map=error_map) 735 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 736 737 deserialized = None 738 if response.status_code == 200: 739 deserialized = self._deserialize('ExpressRouteCircuitsRoutesTableListResult', pipeline_response) 740 741 if cls: 742 return cls(pipeline_response, deserialized, {}) 743 744 return deserialized 745 _list_routes_table_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}/peerings/{peeringName}/routeTables/{devicePath}'} # type: ignore 746 747 async def begin_list_routes_table( 748 self, 749 resource_group_name: str, 750 cross_connection_name: str, 751 peering_name: str, 752 device_path: str, 753 **kwargs 754 ) -> AsyncLROPoller["_models.ExpressRouteCircuitsRoutesTableListResult"]: 755 """Gets the currently advertised routes table associated with the express route cross connection 756 in a resource group. 757 758 :param resource_group_name: The name of the resource group. 759 :type resource_group_name: str 760 :param cross_connection_name: The name of the ExpressRouteCrossConnection. 761 :type cross_connection_name: str 762 :param peering_name: The name of the peering. 763 :type peering_name: str 764 :param device_path: The path of the device. 765 :type device_path: str 766 :keyword callable cls: A custom type or function that will be passed the direct response 767 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 768 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 769 False for no polling, or your own initialized polling object for a personal polling strategy. 770 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 771 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 772 :return: An instance of AsyncLROPoller that returns either ExpressRouteCircuitsRoutesTableListResult or the result of cls(response) 773 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_06_01.models.ExpressRouteCircuitsRoutesTableListResult] 774 :raises ~azure.core.exceptions.HttpResponseError: 775 """ 776 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 777 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteCircuitsRoutesTableListResult"] 778 lro_delay = kwargs.pop( 779 'polling_interval', 780 self._config.polling_interval 781 ) 782 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 783 if cont_token is None: 784 raw_result = await self._list_routes_table_initial( 785 resource_group_name=resource_group_name, 786 cross_connection_name=cross_connection_name, 787 peering_name=peering_name, 788 device_path=device_path, 789 cls=lambda x,y,z: x, 790 **kwargs 791 ) 792 793 kwargs.pop('error_map', None) 794 kwargs.pop('content_type', None) 795 796 def get_long_running_output(pipeline_response): 797 deserialized = self._deserialize('ExpressRouteCircuitsRoutesTableListResult', pipeline_response) 798 799 if cls: 800 return cls(pipeline_response, deserialized, {}) 801 return deserialized 802 803 path_format_arguments = { 804 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 805 'crossConnectionName': self._serialize.url("cross_connection_name", cross_connection_name, 'str'), 806 'peeringName': self._serialize.url("peering_name", peering_name, 'str'), 807 'devicePath': self._serialize.url("device_path", device_path, 'str'), 808 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 809 } 810 811 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 812 elif polling is False: polling_method = AsyncNoPolling() 813 else: polling_method = polling 814 if cont_token: 815 return AsyncLROPoller.from_continuation_token( 816 polling_method=polling_method, 817 continuation_token=cont_token, 818 client=self._client, 819 deserialization_callback=get_long_running_output 820 ) 821 else: 822 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 823 begin_list_routes_table.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteCrossConnections/{crossConnectionName}/peerings/{peeringName}/routeTables/{devicePath}'} # type: ignore 824