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