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