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
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.mgmt.core.exceptions import ARMErrorFormat
16
17from ... import models as _models
18
19T = TypeVar('T')
20ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]]
21
22class SubscriptionsOperations:
23    """SubscriptionsOperations async operations.
24
25    You should not instantiate this class directly. Instead, you should create a Client instance that
26    instantiates it for you and attaches it as an attribute.
27
28    :ivar models: Alias to model classes used in this operation group.
29    :type models: ~azure.mgmt.resource.subscriptions.v2016_06_01.models
30    :param client: Client for service requests.
31    :param config: Configuration of service client.
32    :param serializer: An object model serializer.
33    :param deserializer: An object model deserializer.
34    """
35
36    models = _models
37
38    def __init__(self, client, config, serializer, deserializer) -> None:
39        self._client = client
40        self._serialize = serializer
41        self._deserialize = deserializer
42        self._config = config
43
44    def list_locations(
45        self,
46        subscription_id: str,
47        **kwargs: Any
48    ) -> AsyncIterable["_models.LocationListResult"]:
49        """Gets all available geo-locations.
50
51        This operation provides all the locations that are available for resource providers; however,
52        each resource provider may support a subset of this list.
53
54        :param subscription_id: The ID of the target subscription.
55        :type subscription_id: str
56        :keyword callable cls: A custom type or function that will be passed the direct response
57        :return: An iterator like instance of either LocationListResult or the result of cls(response)
58        :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.subscriptions.v2016_06_01.models.LocationListResult]
59        :raises: ~azure.core.exceptions.HttpResponseError
60        """
61        cls = kwargs.pop('cls', None)  # type: ClsType["_models.LocationListResult"]
62        error_map = {
63            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
64        }
65        error_map.update(kwargs.pop('error_map', {}))
66        api_version = "2016-06-01"
67        accept = "application/json"
68
69        def prepare_request(next_link=None):
70            # Construct headers
71            header_parameters = {}  # type: Dict[str, Any]
72            header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
73
74            if not next_link:
75                # Construct URL
76                url = self.list_locations.metadata['url']  # type: ignore
77                path_format_arguments = {
78                    'subscriptionId': self._serialize.url("subscription_id", subscription_id, 'str'),
79                }
80                url = self._client.format_url(url, **path_format_arguments)
81                # Construct parameters
82                query_parameters = {}  # type: Dict[str, Any]
83                query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
84
85                request = self._client.get(url, query_parameters, header_parameters)
86            else:
87                url = next_link
88                query_parameters = {}  # type: Dict[str, Any]
89                request = self._client.get(url, query_parameters, header_parameters)
90            return request
91
92        async def extract_data(pipeline_response):
93            deserialized = self._deserialize('LocationListResult', pipeline_response)
94            list_of_elem = deserialized.value
95            if cls:
96                list_of_elem = cls(list_of_elem)
97            return None, AsyncList(list_of_elem)
98
99        async def get_next(next_link=None):
100            request = prepare_request(next_link)
101
102            pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
103            response = pipeline_response.http_response
104
105            if response.status_code not in [200]:
106                map_error(status_code=response.status_code, response=response, error_map=error_map)
107                raise HttpResponseError(response=response, error_format=ARMErrorFormat)
108
109            return pipeline_response
110
111        return AsyncItemPaged(
112            get_next, extract_data
113        )
114    list_locations.metadata = {'url': '/subscriptions/{subscriptionId}/locations'}  # type: ignore
115
116    async def get(
117        self,
118        subscription_id: str,
119        **kwargs: Any
120    ) -> "_models.Subscription":
121        """Gets details about a specified subscription.
122
123        :param subscription_id: The ID of the target subscription.
124        :type subscription_id: str
125        :keyword callable cls: A custom type or function that will be passed the direct response
126        :return: Subscription, or the result of cls(response)
127        :rtype: ~azure.mgmt.resource.subscriptions.v2016_06_01.models.Subscription
128        :raises: ~azure.core.exceptions.HttpResponseError
129        """
130        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Subscription"]
131        error_map = {
132            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
133        }
134        error_map.update(kwargs.pop('error_map', {}))
135        api_version = "2016-06-01"
136        accept = "application/json"
137
138        # Construct URL
139        url = self.get.metadata['url']  # type: ignore
140        path_format_arguments = {
141            'subscriptionId': self._serialize.url("subscription_id", subscription_id, 'str'),
142        }
143        url = self._client.format_url(url, **path_format_arguments)
144
145        # Construct parameters
146        query_parameters = {}  # type: Dict[str, Any]
147        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
148
149        # Construct headers
150        header_parameters = {}  # type: Dict[str, Any]
151        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
152
153        request = self._client.get(url, query_parameters, header_parameters)
154        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
155        response = pipeline_response.http_response
156
157        if response.status_code not in [200]:
158            map_error(status_code=response.status_code, response=response, error_map=error_map)
159            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
160
161        deserialized = self._deserialize('Subscription', pipeline_response)
162
163        if cls:
164            return cls(pipeline_response, deserialized, {})
165
166        return deserialized
167    get.metadata = {'url': '/subscriptions/{subscriptionId}'}  # type: ignore
168
169    def list(
170        self,
171        **kwargs: Any
172    ) -> AsyncIterable["_models.SubscriptionListResult"]:
173        """Gets all subscriptions for a tenant.
174
175        :keyword callable cls: A custom type or function that will be passed the direct response
176        :return: An iterator like instance of either SubscriptionListResult or the result of cls(response)
177        :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.subscriptions.v2016_06_01.models.SubscriptionListResult]
178        :raises: ~azure.core.exceptions.HttpResponseError
179        """
180        cls = kwargs.pop('cls', None)  # type: ClsType["_models.SubscriptionListResult"]
181        error_map = {
182            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
183        }
184        error_map.update(kwargs.pop('error_map', {}))
185        api_version = "2016-06-01"
186        accept = "application/json"
187
188        def prepare_request(next_link=None):
189            # Construct headers
190            header_parameters = {}  # type: Dict[str, Any]
191            header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
192
193            if not next_link:
194                # Construct URL
195                url = self.list.metadata['url']  # type: ignore
196                # Construct parameters
197                query_parameters = {}  # type: Dict[str, Any]
198                query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
199
200                request = self._client.get(url, query_parameters, header_parameters)
201            else:
202                url = next_link
203                query_parameters = {}  # type: Dict[str, Any]
204                request = self._client.get(url, query_parameters, header_parameters)
205            return request
206
207        async def extract_data(pipeline_response):
208            deserialized = self._deserialize('SubscriptionListResult', pipeline_response)
209            list_of_elem = deserialized.value
210            if cls:
211                list_of_elem = cls(list_of_elem)
212            return deserialized.next_link or None, AsyncList(list_of_elem)
213
214        async def get_next(next_link=None):
215            request = prepare_request(next_link)
216
217            pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
218            response = pipeline_response.http_response
219
220            if response.status_code not in [200]:
221                map_error(status_code=response.status_code, response=response, error_map=error_map)
222                raise HttpResponseError(response=response, error_format=ARMErrorFormat)
223
224            return pipeline_response
225
226        return AsyncItemPaged(
227            get_next, extract_data
228        )
229    list.metadata = {'url': '/subscriptions'}  # type: ignore
230