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