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, Callable, Dict, Generic, Optional, TypeVar, Union 9import warnings 10 11from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error 12from azure.core.pipeline import PipelineResponse 13from azure.core.pipeline.transport import AsyncHttpResponse, HttpRequest 14from azure.core.polling import AsyncLROPoller, AsyncNoPolling, AsyncPollingMethod 15from azure.mgmt.core.exceptions import ARMErrorFormat 16from azure.mgmt.core.polling.async_arm_polling import AsyncARMPolling 17 18from ... import models as _models 19 20T = TypeVar('T') 21ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]] 22 23class ExpressRouteConnectionsOperations: 24 """ExpressRouteConnectionsOperations async operations. 25 26 You should not instantiate this class directly. Instead, you should create a Client instance that 27 instantiates it for you and attaches it as an attribute. 28 29 :ivar models: Alias to model classes used in this operation group. 30 :type models: ~azure.mgmt.network.v2020_03_01.models 31 :param client: Client for service requests. 32 :param config: Configuration of service client. 33 :param serializer: An object model serializer. 34 :param deserializer: An object model deserializer. 35 """ 36 37 models = _models 38 39 def __init__(self, client, config, serializer, deserializer) -> None: 40 self._client = client 41 self._serialize = serializer 42 self._deserialize = deserializer 43 self._config = config 44 45 async def _create_or_update_initial( 46 self, 47 resource_group_name: str, 48 express_route_gateway_name: str, 49 connection_name: str, 50 put_express_route_connection_parameters: "_models.ExpressRouteConnection", 51 **kwargs 52 ) -> "_models.ExpressRouteConnection": 53 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteConnection"] 54 error_map = { 55 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 56 } 57 error_map.update(kwargs.pop('error_map', {})) 58 api_version = "2020-03-01" 59 content_type = kwargs.pop("content_type", "application/json") 60 accept = "application/json" 61 62 # Construct URL 63 url = self._create_or_update_initial.metadata['url'] # type: ignore 64 path_format_arguments = { 65 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 66 'expressRouteGatewayName': self._serialize.url("express_route_gateway_name", express_route_gateway_name, 'str'), 67 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 68 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 69 } 70 url = self._client.format_url(url, **path_format_arguments) 71 72 # Construct parameters 73 query_parameters = {} # type: Dict[str, Any] 74 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 75 76 # Construct headers 77 header_parameters = {} # type: Dict[str, Any] 78 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 79 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 80 81 body_content_kwargs = {} # type: Dict[str, Any] 82 body_content = self._serialize.body(put_express_route_connection_parameters, 'ExpressRouteConnection') 83 body_content_kwargs['content'] = body_content 84 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 85 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 86 response = pipeline_response.http_response 87 88 if response.status_code not in [200, 201]: 89 map_error(status_code=response.status_code, response=response, error_map=error_map) 90 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 91 92 if response.status_code == 200: 93 deserialized = self._deserialize('ExpressRouteConnection', pipeline_response) 94 95 if response.status_code == 201: 96 deserialized = self._deserialize('ExpressRouteConnection', pipeline_response) 97 98 if cls: 99 return cls(pipeline_response, deserialized, {}) 100 101 return deserialized 102 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteGateways/{expressRouteGatewayName}/expressRouteConnections/{connectionName}'} # type: ignore 103 104 async def begin_create_or_update( 105 self, 106 resource_group_name: str, 107 express_route_gateway_name: str, 108 connection_name: str, 109 put_express_route_connection_parameters: "_models.ExpressRouteConnection", 110 **kwargs 111 ) -> AsyncLROPoller["_models.ExpressRouteConnection"]: 112 """Creates a connection between an ExpressRoute gateway and an ExpressRoute circuit. 113 114 :param resource_group_name: The name of the resource group. 115 :type resource_group_name: str 116 :param express_route_gateway_name: The name of the ExpressRoute gateway. 117 :type express_route_gateway_name: str 118 :param connection_name: The name of the connection subresource. 119 :type connection_name: str 120 :param put_express_route_connection_parameters: Parameters required in an 121 ExpressRouteConnection PUT operation. 122 :type put_express_route_connection_parameters: ~azure.mgmt.network.v2020_03_01.models.ExpressRouteConnection 123 :keyword callable cls: A custom type or function that will be passed the direct response 124 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 125 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 126 False for no polling, or your own initialized polling object for a personal polling strategy. 127 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 128 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 129 :return: An instance of AsyncLROPoller that returns either ExpressRouteConnection or the result of cls(response) 130 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_03_01.models.ExpressRouteConnection] 131 :raises ~azure.core.exceptions.HttpResponseError: 132 """ 133 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 134 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteConnection"] 135 lro_delay = kwargs.pop( 136 'polling_interval', 137 self._config.polling_interval 138 ) 139 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 140 if cont_token is None: 141 raw_result = await self._create_or_update_initial( 142 resource_group_name=resource_group_name, 143 express_route_gateway_name=express_route_gateway_name, 144 connection_name=connection_name, 145 put_express_route_connection_parameters=put_express_route_connection_parameters, 146 cls=lambda x,y,z: x, 147 **kwargs 148 ) 149 150 kwargs.pop('error_map', None) 151 kwargs.pop('content_type', None) 152 153 def get_long_running_output(pipeline_response): 154 deserialized = self._deserialize('ExpressRouteConnection', pipeline_response) 155 156 if cls: 157 return cls(pipeline_response, deserialized, {}) 158 return deserialized 159 160 path_format_arguments = { 161 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 162 'expressRouteGatewayName': self._serialize.url("express_route_gateway_name", express_route_gateway_name, 'str'), 163 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 164 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 165 } 166 167 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 168 elif polling is False: polling_method = AsyncNoPolling() 169 else: polling_method = polling 170 if cont_token: 171 return AsyncLROPoller.from_continuation_token( 172 polling_method=polling_method, 173 continuation_token=cont_token, 174 client=self._client, 175 deserialization_callback=get_long_running_output 176 ) 177 else: 178 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 179 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteGateways/{expressRouteGatewayName}/expressRouteConnections/{connectionName}'} # type: ignore 180 181 async def get( 182 self, 183 resource_group_name: str, 184 express_route_gateway_name: str, 185 connection_name: str, 186 **kwargs 187 ) -> "_models.ExpressRouteConnection": 188 """Gets the specified ExpressRouteConnection. 189 190 :param resource_group_name: The name of the resource group. 191 :type resource_group_name: str 192 :param express_route_gateway_name: The name of the ExpressRoute gateway. 193 :type express_route_gateway_name: str 194 :param connection_name: The name of the ExpressRoute connection. 195 :type connection_name: str 196 :keyword callable cls: A custom type or function that will be passed the direct response 197 :return: ExpressRouteConnection, or the result of cls(response) 198 :rtype: ~azure.mgmt.network.v2020_03_01.models.ExpressRouteConnection 199 :raises: ~azure.core.exceptions.HttpResponseError 200 """ 201 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteConnection"] 202 error_map = { 203 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 204 } 205 error_map.update(kwargs.pop('error_map', {})) 206 api_version = "2020-03-01" 207 accept = "application/json" 208 209 # Construct URL 210 url = self.get.metadata['url'] # type: ignore 211 path_format_arguments = { 212 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 213 'expressRouteGatewayName': self._serialize.url("express_route_gateway_name", express_route_gateway_name, 'str'), 214 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 215 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 216 } 217 url = self._client.format_url(url, **path_format_arguments) 218 219 # Construct parameters 220 query_parameters = {} # type: Dict[str, Any] 221 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 222 223 # Construct headers 224 header_parameters = {} # type: Dict[str, Any] 225 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 226 227 request = self._client.get(url, query_parameters, header_parameters) 228 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 229 response = pipeline_response.http_response 230 231 if response.status_code not in [200]: 232 map_error(status_code=response.status_code, response=response, error_map=error_map) 233 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 234 235 deserialized = self._deserialize('ExpressRouteConnection', pipeline_response) 236 237 if cls: 238 return cls(pipeline_response, deserialized, {}) 239 240 return deserialized 241 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteGateways/{expressRouteGatewayName}/expressRouteConnections/{connectionName}'} # type: ignore 242 243 async def _delete_initial( 244 self, 245 resource_group_name: str, 246 express_route_gateway_name: str, 247 connection_name: str, 248 **kwargs 249 ) -> None: 250 cls = kwargs.pop('cls', None) # type: ClsType[None] 251 error_map = { 252 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 253 } 254 error_map.update(kwargs.pop('error_map', {})) 255 api_version = "2020-03-01" 256 accept = "application/json" 257 258 # Construct URL 259 url = self._delete_initial.metadata['url'] # type: ignore 260 path_format_arguments = { 261 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 262 'expressRouteGatewayName': self._serialize.url("express_route_gateway_name", express_route_gateway_name, 'str'), 263 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 264 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 265 } 266 url = self._client.format_url(url, **path_format_arguments) 267 268 # Construct parameters 269 query_parameters = {} # type: Dict[str, Any] 270 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 271 272 # Construct headers 273 header_parameters = {} # type: Dict[str, Any] 274 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 275 276 request = self._client.delete(url, query_parameters, header_parameters) 277 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 278 response = pipeline_response.http_response 279 280 if response.status_code not in [200, 202, 204]: 281 map_error(status_code=response.status_code, response=response, error_map=error_map) 282 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 283 284 if cls: 285 return cls(pipeline_response, None, {}) 286 287 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteGateways/{expressRouteGatewayName}/expressRouteConnections/{connectionName}'} # type: ignore 288 289 async def begin_delete( 290 self, 291 resource_group_name: str, 292 express_route_gateway_name: str, 293 connection_name: str, 294 **kwargs 295 ) -> AsyncLROPoller[None]: 296 """Deletes a connection to a ExpressRoute circuit. 297 298 :param resource_group_name: The name of the resource group. 299 :type resource_group_name: str 300 :param express_route_gateway_name: The name of the ExpressRoute gateway. 301 :type express_route_gateway_name: str 302 :param connection_name: The name of the connection subresource. 303 :type connection_name: str 304 :keyword callable cls: A custom type or function that will be passed the direct response 305 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 306 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 307 False for no polling, or your own initialized polling object for a personal polling strategy. 308 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 309 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 310 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 311 :rtype: ~azure.core.polling.AsyncLROPoller[None] 312 :raises ~azure.core.exceptions.HttpResponseError: 313 """ 314 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 315 cls = kwargs.pop('cls', None) # type: ClsType[None] 316 lro_delay = kwargs.pop( 317 'polling_interval', 318 self._config.polling_interval 319 ) 320 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 321 if cont_token is None: 322 raw_result = await self._delete_initial( 323 resource_group_name=resource_group_name, 324 express_route_gateway_name=express_route_gateway_name, 325 connection_name=connection_name, 326 cls=lambda x,y,z: x, 327 **kwargs 328 ) 329 330 kwargs.pop('error_map', None) 331 kwargs.pop('content_type', None) 332 333 def get_long_running_output(pipeline_response): 334 if cls: 335 return cls(pipeline_response, None, {}) 336 337 path_format_arguments = { 338 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 339 'expressRouteGatewayName': self._serialize.url("express_route_gateway_name", express_route_gateway_name, 'str'), 340 'connectionName': self._serialize.url("connection_name", connection_name, 'str'), 341 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 342 } 343 344 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 345 elif polling is False: polling_method = AsyncNoPolling() 346 else: polling_method = polling 347 if cont_token: 348 return AsyncLROPoller.from_continuation_token( 349 polling_method=polling_method, 350 continuation_token=cont_token, 351 client=self._client, 352 deserialization_callback=get_long_running_output 353 ) 354 else: 355 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 356 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteGateways/{expressRouteGatewayName}/expressRouteConnections/{connectionName}'} # type: ignore 357 358 async def list( 359 self, 360 resource_group_name: str, 361 express_route_gateway_name: str, 362 **kwargs 363 ) -> "_models.ExpressRouteConnectionList": 364 """Lists ExpressRouteConnections. 365 366 :param resource_group_name: The name of the resource group. 367 :type resource_group_name: str 368 :param express_route_gateway_name: The name of the ExpressRoute gateway. 369 :type express_route_gateway_name: str 370 :keyword callable cls: A custom type or function that will be passed the direct response 371 :return: ExpressRouteConnectionList, or the result of cls(response) 372 :rtype: ~azure.mgmt.network.v2020_03_01.models.ExpressRouteConnectionList 373 :raises: ~azure.core.exceptions.HttpResponseError 374 """ 375 cls = kwargs.pop('cls', None) # type: ClsType["_models.ExpressRouteConnectionList"] 376 error_map = { 377 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 378 } 379 error_map.update(kwargs.pop('error_map', {})) 380 api_version = "2020-03-01" 381 accept = "application/json" 382 383 # Construct URL 384 url = self.list.metadata['url'] # type: ignore 385 path_format_arguments = { 386 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 387 'expressRouteGatewayName': self._serialize.url("express_route_gateway_name", express_route_gateway_name, 'str'), 388 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 389 } 390 url = self._client.format_url(url, **path_format_arguments) 391 392 # Construct parameters 393 query_parameters = {} # type: Dict[str, Any] 394 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 395 396 # Construct headers 397 header_parameters = {} # type: Dict[str, Any] 398 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 399 400 request = self._client.get(url, query_parameters, header_parameters) 401 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 402 response = pipeline_response.http_response 403 404 if response.status_code not in [200]: 405 map_error(status_code=response.status_code, response=response, error_map=error_map) 406 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 407 408 deserialized = self._deserialize('ExpressRouteConnectionList', pipeline_response) 409 410 if cls: 411 return cls(pipeline_response, deserialized, {}) 412 413 return deserialized 414 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/expressRouteGateways/{expressRouteGatewayName}/expressRouteConnections'} # type: ignore 415