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 ProvidersOperations:
23    """ProvidersOperations 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.resources.v2020_10_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    async def unregister(
45        self,
46        resource_provider_namespace: str,
47        **kwargs: Any
48    ) -> "_models.Provider":
49        """Unregisters a subscription from a resource provider.
50
51        :param resource_provider_namespace: The namespace of the resource provider to unregister.
52        :type resource_provider_namespace: str
53        :keyword callable cls: A custom type or function that will be passed the direct response
54        :return: Provider, or the result of cls(response)
55        :rtype: ~azure.mgmt.resource.resources.v2020_10_01.models.Provider
56        :raises: ~azure.core.exceptions.HttpResponseError
57        """
58        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Provider"]
59        error_map = {
60            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
61        }
62        error_map.update(kwargs.pop('error_map', {}))
63        api_version = "2020-10-01"
64        accept = "application/json"
65
66        # Construct URL
67        url = self.unregister.metadata['url']  # type: ignore
68        path_format_arguments = {
69            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
70            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
71        }
72        url = self._client.format_url(url, **path_format_arguments)
73
74        # Construct parameters
75        query_parameters = {}  # type: Dict[str, Any]
76        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
77
78        # Construct headers
79        header_parameters = {}  # type: Dict[str, Any]
80        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
81
82        request = self._client.post(url, query_parameters, header_parameters)
83        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
84        response = pipeline_response.http_response
85
86        if response.status_code not in [200]:
87            map_error(status_code=response.status_code, response=response, error_map=error_map)
88            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
89
90        deserialized = self._deserialize('Provider', pipeline_response)
91
92        if cls:
93            return cls(pipeline_response, deserialized, {})
94
95        return deserialized
96    unregister.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}/unregister'}  # type: ignore
97
98    async def register_at_management_group_scope(
99        self,
100        resource_provider_namespace: str,
101        group_id: str,
102        **kwargs: Any
103    ) -> None:
104        """Registers a management group with a resource provider.
105
106        :param resource_provider_namespace: The namespace of the resource provider to register.
107        :type resource_provider_namespace: str
108        :param group_id: The management group ID.
109        :type group_id: str
110        :keyword callable cls: A custom type or function that will be passed the direct response
111        :return: None, or the result of cls(response)
112        :rtype: None
113        :raises: ~azure.core.exceptions.HttpResponseError
114        """
115        cls = kwargs.pop('cls', None)  # type: ClsType[None]
116        error_map = {
117            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
118        }
119        error_map.update(kwargs.pop('error_map', {}))
120        api_version = "2020-10-01"
121        accept = "application/json"
122
123        # Construct URL
124        url = self.register_at_management_group_scope.metadata['url']  # type: ignore
125        path_format_arguments = {
126            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
127            'groupId': self._serialize.url("group_id", group_id, 'str', max_length=90, min_length=1),
128        }
129        url = self._client.format_url(url, **path_format_arguments)
130
131        # Construct parameters
132        query_parameters = {}  # type: Dict[str, Any]
133        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
134
135        # Construct headers
136        header_parameters = {}  # type: Dict[str, Any]
137        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
138
139        request = self._client.post(url, query_parameters, header_parameters)
140        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
141        response = pipeline_response.http_response
142
143        if response.status_code not in [200]:
144            map_error(status_code=response.status_code, response=response, error_map=error_map)
145            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
146
147        if cls:
148            return cls(pipeline_response, None, {})
149
150    register_at_management_group_scope.metadata = {'url': '/providers/Microsoft.Management/managementGroups/{groupId}/providers/{resourceProviderNamespace}/register'}  # type: ignore
151
152    async def register(
153        self,
154        resource_provider_namespace: str,
155        **kwargs: Any
156    ) -> "_models.Provider":
157        """Registers a subscription with a resource provider.
158
159        :param resource_provider_namespace: The namespace of the resource provider to register.
160        :type resource_provider_namespace: str
161        :keyword callable cls: A custom type or function that will be passed the direct response
162        :return: Provider, or the result of cls(response)
163        :rtype: ~azure.mgmt.resource.resources.v2020_10_01.models.Provider
164        :raises: ~azure.core.exceptions.HttpResponseError
165        """
166        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Provider"]
167        error_map = {
168            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
169        }
170        error_map.update(kwargs.pop('error_map', {}))
171        api_version = "2020-10-01"
172        accept = "application/json"
173
174        # Construct URL
175        url = self.register.metadata['url']  # type: ignore
176        path_format_arguments = {
177            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
178            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
179        }
180        url = self._client.format_url(url, **path_format_arguments)
181
182        # Construct parameters
183        query_parameters = {}  # type: Dict[str, Any]
184        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
185
186        # Construct headers
187        header_parameters = {}  # type: Dict[str, Any]
188        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
189
190        request = self._client.post(url, query_parameters, header_parameters)
191        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
192        response = pipeline_response.http_response
193
194        if response.status_code not in [200]:
195            map_error(status_code=response.status_code, response=response, error_map=error_map)
196            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
197
198        deserialized = self._deserialize('Provider', pipeline_response)
199
200        if cls:
201            return cls(pipeline_response, deserialized, {})
202
203        return deserialized
204    register.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}/register'}  # type: ignore
205
206    def list(
207        self,
208        top: Optional[int] = None,
209        expand: Optional[str] = None,
210        **kwargs: Any
211    ) -> AsyncIterable["_models.ProviderListResult"]:
212        """Gets all resource providers for a subscription.
213
214        :param top: The number of results to return. If null is passed returns all deployments.
215        :type top: int
216        :param expand: The properties to include in the results. For example, use &$expand=metadata in
217         the query string to retrieve resource provider metadata. To include property aliases in
218         response, use $expand=resourceTypes/aliases.
219        :type expand: str
220        :keyword callable cls: A custom type or function that will be passed the direct response
221        :return: An iterator like instance of either ProviderListResult or the result of cls(response)
222        :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.resources.v2020_10_01.models.ProviderListResult]
223        :raises: ~azure.core.exceptions.HttpResponseError
224        """
225        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ProviderListResult"]
226        error_map = {
227            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
228        }
229        error_map.update(kwargs.pop('error_map', {}))
230        api_version = "2020-10-01"
231        accept = "application/json"
232
233        def prepare_request(next_link=None):
234            # Construct headers
235            header_parameters = {}  # type: Dict[str, Any]
236            header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
237
238            if not next_link:
239                # Construct URL
240                url = self.list.metadata['url']  # type: ignore
241                path_format_arguments = {
242                    'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
243                }
244                url = self._client.format_url(url, **path_format_arguments)
245                # Construct parameters
246                query_parameters = {}  # type: Dict[str, Any]
247                if top is not None:
248                    query_parameters['$top'] = self._serialize.query("top", top, 'int')
249                if expand is not None:
250                    query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
251                query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
252
253                request = self._client.get(url, query_parameters, header_parameters)
254            else:
255                url = next_link
256                query_parameters = {}  # type: Dict[str, Any]
257                request = self._client.get(url, query_parameters, header_parameters)
258            return request
259
260        async def extract_data(pipeline_response):
261            deserialized = self._deserialize('ProviderListResult', pipeline_response)
262            list_of_elem = deserialized.value
263            if cls:
264                list_of_elem = cls(list_of_elem)
265            return deserialized.next_link or None, AsyncList(list_of_elem)
266
267        async def get_next(next_link=None):
268            request = prepare_request(next_link)
269
270            pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
271            response = pipeline_response.http_response
272
273            if response.status_code not in [200]:
274                map_error(status_code=response.status_code, response=response, error_map=error_map)
275                raise HttpResponseError(response=response, error_format=ARMErrorFormat)
276
277            return pipeline_response
278
279        return AsyncItemPaged(
280            get_next, extract_data
281        )
282    list.metadata = {'url': '/subscriptions/{subscriptionId}/providers'}  # type: ignore
283
284    def list_at_tenant_scope(
285        self,
286        top: Optional[int] = None,
287        expand: Optional[str] = None,
288        **kwargs: Any
289    ) -> AsyncIterable["_models.ProviderListResult"]:
290        """Gets all resource providers for the tenant.
291
292        :param top: The number of results to return. If null is passed returns all providers.
293        :type top: int
294        :param expand: The properties to include in the results. For example, use &$expand=metadata in
295         the query string to retrieve resource provider metadata. To include property aliases in
296         response, use $expand=resourceTypes/aliases.
297        :type expand: str
298        :keyword callable cls: A custom type or function that will be passed the direct response
299        :return: An iterator like instance of either ProviderListResult or the result of cls(response)
300        :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.resources.v2020_10_01.models.ProviderListResult]
301        :raises: ~azure.core.exceptions.HttpResponseError
302        """
303        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ProviderListResult"]
304        error_map = {
305            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
306        }
307        error_map.update(kwargs.pop('error_map', {}))
308        api_version = "2020-10-01"
309        accept = "application/json"
310
311        def prepare_request(next_link=None):
312            # Construct headers
313            header_parameters = {}  # type: Dict[str, Any]
314            header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
315
316            if not next_link:
317                # Construct URL
318                url = self.list_at_tenant_scope.metadata['url']  # type: ignore
319                # Construct parameters
320                query_parameters = {}  # type: Dict[str, Any]
321                if top is not None:
322                    query_parameters['$top'] = self._serialize.query("top", top, 'int')
323                if expand is not None:
324                    query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
325                query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
326
327                request = self._client.get(url, query_parameters, header_parameters)
328            else:
329                url = next_link
330                query_parameters = {}  # type: Dict[str, Any]
331                request = self._client.get(url, query_parameters, header_parameters)
332            return request
333
334        async def extract_data(pipeline_response):
335            deserialized = self._deserialize('ProviderListResult', pipeline_response)
336            list_of_elem = deserialized.value
337            if cls:
338                list_of_elem = cls(list_of_elem)
339            return deserialized.next_link or None, AsyncList(list_of_elem)
340
341        async def get_next(next_link=None):
342            request = prepare_request(next_link)
343
344            pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
345            response = pipeline_response.http_response
346
347            if response.status_code not in [200]:
348                map_error(status_code=response.status_code, response=response, error_map=error_map)
349                raise HttpResponseError(response=response, error_format=ARMErrorFormat)
350
351            return pipeline_response
352
353        return AsyncItemPaged(
354            get_next, extract_data
355        )
356    list_at_tenant_scope.metadata = {'url': '/providers'}  # type: ignore
357
358    async def get(
359        self,
360        resource_provider_namespace: str,
361        expand: Optional[str] = None,
362        **kwargs: Any
363    ) -> "_models.Provider":
364        """Gets the specified resource provider.
365
366        :param resource_provider_namespace: The namespace of the resource provider.
367        :type resource_provider_namespace: str
368        :param expand: The $expand query parameter. For example, to include property aliases in
369         response, use $expand=resourceTypes/aliases.
370        :type expand: str
371        :keyword callable cls: A custom type or function that will be passed the direct response
372        :return: Provider, or the result of cls(response)
373        :rtype: ~azure.mgmt.resource.resources.v2020_10_01.models.Provider
374        :raises: ~azure.core.exceptions.HttpResponseError
375        """
376        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Provider"]
377        error_map = {
378            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
379        }
380        error_map.update(kwargs.pop('error_map', {}))
381        api_version = "2020-10-01"
382        accept = "application/json"
383
384        # Construct URL
385        url = self.get.metadata['url']  # type: ignore
386        path_format_arguments = {
387            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
388            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
389        }
390        url = self._client.format_url(url, **path_format_arguments)
391
392        # Construct parameters
393        query_parameters = {}  # type: Dict[str, Any]
394        if expand is not None:
395            query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
396        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
397
398        # Construct headers
399        header_parameters = {}  # type: Dict[str, Any]
400        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
401
402        request = self._client.get(url, query_parameters, header_parameters)
403        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
404        response = pipeline_response.http_response
405
406        if response.status_code not in [200]:
407            map_error(status_code=response.status_code, response=response, error_map=error_map)
408            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
409
410        deserialized = self._deserialize('Provider', pipeline_response)
411
412        if cls:
413            return cls(pipeline_response, deserialized, {})
414
415        return deserialized
416    get.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}'}  # type: ignore
417
418    async def get_at_tenant_scope(
419        self,
420        resource_provider_namespace: str,
421        expand: Optional[str] = None,
422        **kwargs: Any
423    ) -> "_models.Provider":
424        """Gets the specified resource provider at the tenant level.
425
426        :param resource_provider_namespace: The namespace of the resource provider.
427        :type resource_provider_namespace: str
428        :param expand: The $expand query parameter. For example, to include property aliases in
429         response, use $expand=resourceTypes/aliases.
430        :type expand: str
431        :keyword callable cls: A custom type or function that will be passed the direct response
432        :return: Provider, or the result of cls(response)
433        :rtype: ~azure.mgmt.resource.resources.v2020_10_01.models.Provider
434        :raises: ~azure.core.exceptions.HttpResponseError
435        """
436        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Provider"]
437        error_map = {
438            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
439        }
440        error_map.update(kwargs.pop('error_map', {}))
441        api_version = "2020-10-01"
442        accept = "application/json"
443
444        # Construct URL
445        url = self.get_at_tenant_scope.metadata['url']  # type: ignore
446        path_format_arguments = {
447            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
448        }
449        url = self._client.format_url(url, **path_format_arguments)
450
451        # Construct parameters
452        query_parameters = {}  # type: Dict[str, Any]
453        if expand is not None:
454            query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
455        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
456
457        # Construct headers
458        header_parameters = {}  # type: Dict[str, Any]
459        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
460
461        request = self._client.get(url, query_parameters, header_parameters)
462        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
463        response = pipeline_response.http_response
464
465        if response.status_code not in [200]:
466            map_error(status_code=response.status_code, response=response, error_map=error_map)
467            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
468
469        deserialized = self._deserialize('Provider', pipeline_response)
470
471        if cls:
472            return cls(pipeline_response, deserialized, {})
473
474        return deserialized
475    get_at_tenant_scope.metadata = {'url': '/providers/{resourceProviderNamespace}'}  # type: ignore
476