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