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