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.mgmt.core.exceptions import ARMErrorFormat
16
17from .. import models as _models
18
19if TYPE_CHECKING:
20    # pylint: disable=unused-import,ungrouped-imports
21    from typing import Any, Callable, Dict, Generic, Iterable, Optional, TypeVar
22
23    T = TypeVar('T')
24    ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
25
26class ProvidersOperations(object):
27    """ProvidersOperations operations.
28
29    You should not instantiate this class directly. Instead, you should create a Client instance that
30    instantiates it for you and attaches it as an attribute.
31
32    :ivar models: Alias to model classes used in this operation group.
33    :type models: ~azure.mgmt.resource.resources.v2021_04_01.models
34    :param client: Client for service requests.
35    :param config: Configuration of service client.
36    :param serializer: An object model serializer.
37    :param deserializer: An object model deserializer.
38    """
39
40    models = _models
41
42    def __init__(self, client, config, serializer, deserializer):
43        self._client = client
44        self._serialize = serializer
45        self._deserialize = deserializer
46        self._config = config
47
48    def unregister(
49        self,
50        resource_provider_namespace,  # type: str
51        **kwargs  # type: Any
52    ):
53        # type: (...) -> "_models.Provider"
54        """Unregisters a subscription from a resource provider.
55
56        :param resource_provider_namespace: The namespace of the resource provider to unregister.
57        :type resource_provider_namespace: str
58        :keyword callable cls: A custom type or function that will be passed the direct response
59        :return: Provider, or the result of cls(response)
60        :rtype: ~azure.mgmt.resource.resources.v2021_04_01.models.Provider
61        :raises: ~azure.core.exceptions.HttpResponseError
62        """
63        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Provider"]
64        error_map = {
65            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
66        }
67        error_map.update(kwargs.pop('error_map', {}))
68        api_version = "2021-04-01"
69        accept = "application/json"
70
71        # Construct URL
72        url = self.unregister.metadata['url']  # type: ignore
73        path_format_arguments = {
74            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
75            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
76        }
77        url = self._client.format_url(url, **path_format_arguments)
78
79        # Construct parameters
80        query_parameters = {}  # type: Dict[str, Any]
81        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
82
83        # Construct headers
84        header_parameters = {}  # type: Dict[str, Any]
85        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
86
87        request = self._client.post(url, query_parameters, header_parameters)
88        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
89        response = pipeline_response.http_response
90
91        if response.status_code not in [200]:
92            map_error(status_code=response.status_code, response=response, error_map=error_map)
93            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
94
95        deserialized = self._deserialize('Provider', pipeline_response)
96
97        if cls:
98            return cls(pipeline_response, deserialized, {})
99
100        return deserialized
101    unregister.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}/unregister'}  # type: ignore
102
103    def register_at_management_group_scope(
104        self,
105        resource_provider_namespace,  # type: str
106        group_id,  # type: str
107        **kwargs  # type: Any
108    ):
109        # type: (...) -> None
110        """Registers a management group with a resource provider.
111
112        :param resource_provider_namespace: The namespace of the resource provider to register.
113        :type resource_provider_namespace: str
114        :param group_id: The management group ID.
115        :type group_id: str
116        :keyword callable cls: A custom type or function that will be passed the direct response
117        :return: None, or the result of cls(response)
118        :rtype: None
119        :raises: ~azure.core.exceptions.HttpResponseError
120        """
121        cls = kwargs.pop('cls', None)  # type: ClsType[None]
122        error_map = {
123            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
124        }
125        error_map.update(kwargs.pop('error_map', {}))
126        api_version = "2021-04-01"
127        accept = "application/json"
128
129        # Construct URL
130        url = self.register_at_management_group_scope.metadata['url']  # type: ignore
131        path_format_arguments = {
132            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
133            'groupId': self._serialize.url("group_id", group_id, 'str', max_length=90, min_length=1),
134        }
135        url = self._client.format_url(url, **path_format_arguments)
136
137        # Construct parameters
138        query_parameters = {}  # type: Dict[str, Any]
139        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
140
141        # Construct headers
142        header_parameters = {}  # type: Dict[str, Any]
143        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
144
145        request = self._client.post(url, query_parameters, header_parameters)
146        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
147        response = pipeline_response.http_response
148
149        if response.status_code not in [200]:
150            map_error(status_code=response.status_code, response=response, error_map=error_map)
151            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
152
153        if cls:
154            return cls(pipeline_response, None, {})
155
156    register_at_management_group_scope.metadata = {'url': '/providers/Microsoft.Management/managementGroups/{groupId}/providers/{resourceProviderNamespace}/register'}  # type: ignore
157
158    def provider_permissions(
159        self,
160        resource_provider_namespace,  # type: str
161        **kwargs  # type: Any
162    ):
163        # type: (...) -> "_models.ProviderPermissionListResult"
164        """Get the provider permissions.
165
166        :param resource_provider_namespace: The namespace of the resource provider.
167        :type resource_provider_namespace: str
168        :keyword callable cls: A custom type or function that will be passed the direct response
169        :return: ProviderPermissionListResult, or the result of cls(response)
170        :rtype: ~azure.mgmt.resource.resources.v2021_04_01.models.ProviderPermissionListResult
171        :raises: ~azure.core.exceptions.HttpResponseError
172        """
173        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ProviderPermissionListResult"]
174        error_map = {
175            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
176        }
177        error_map.update(kwargs.pop('error_map', {}))
178        api_version = "2021-04-01"
179        accept = "application/json"
180
181        # Construct URL
182        url = self.provider_permissions.metadata['url']  # type: ignore
183        path_format_arguments = {
184            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
185            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
186        }
187        url = self._client.format_url(url, **path_format_arguments)
188
189        # Construct parameters
190        query_parameters = {}  # type: Dict[str, Any]
191        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
192
193        # Construct headers
194        header_parameters = {}  # type: Dict[str, Any]
195        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
196
197        request = self._client.get(url, query_parameters, header_parameters)
198        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
199        response = pipeline_response.http_response
200
201        if response.status_code not in [200]:
202            map_error(status_code=response.status_code, response=response, error_map=error_map)
203            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
204
205        deserialized = self._deserialize('ProviderPermissionListResult', pipeline_response)
206
207        if cls:
208            return cls(pipeline_response, deserialized, {})
209
210        return deserialized
211    provider_permissions.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}/providerPermissions'}  # type: ignore
212
213    def register(
214        self,
215        resource_provider_namespace,  # type: str
216        properties=None,  # type: Optional["_models.ProviderRegistrationRequest"]
217        **kwargs  # type: Any
218    ):
219        # type: (...) -> "_models.Provider"
220        """Registers a subscription with a resource provider.
221
222        :param resource_provider_namespace: The namespace of the resource provider to register.
223        :type resource_provider_namespace: str
224        :param properties: The third party consent for S2S.
225        :type properties: ~azure.mgmt.resource.resources.v2021_04_01.models.ProviderRegistrationRequest
226        :keyword callable cls: A custom type or function that will be passed the direct response
227        :return: Provider, or the result of cls(response)
228        :rtype: ~azure.mgmt.resource.resources.v2021_04_01.models.Provider
229        :raises: ~azure.core.exceptions.HttpResponseError
230        """
231        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Provider"]
232        error_map = {
233            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
234        }
235        error_map.update(kwargs.pop('error_map', {}))
236        api_version = "2021-04-01"
237        content_type = kwargs.pop("content_type", "application/json")
238        accept = "application/json"
239
240        # Construct URL
241        url = self.register.metadata['url']  # type: ignore
242        path_format_arguments = {
243            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
244            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
245        }
246        url = self._client.format_url(url, **path_format_arguments)
247
248        # Construct parameters
249        query_parameters = {}  # type: Dict[str, Any]
250        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
251
252        # Construct headers
253        header_parameters = {}  # type: Dict[str, Any]
254        header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str')
255        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
256
257        body_content_kwargs = {}  # type: Dict[str, Any]
258        if properties is not None:
259            body_content = self._serialize.body(properties, 'ProviderRegistrationRequest')
260        else:
261            body_content = None
262        body_content_kwargs['content'] = body_content
263        request = self._client.post(url, query_parameters, header_parameters, **body_content_kwargs)
264        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
265        response = pipeline_response.http_response
266
267        if response.status_code not in [200]:
268            map_error(status_code=response.status_code, response=response, error_map=error_map)
269            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
270
271        deserialized = self._deserialize('Provider', pipeline_response)
272
273        if cls:
274            return cls(pipeline_response, deserialized, {})
275
276        return deserialized
277    register.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}/register'}  # type: ignore
278
279    def list(
280        self,
281        top=None,  # type: Optional[int]
282        expand=None,  # type: Optional[str]
283        **kwargs  # type: Any
284    ):
285        # type: (...) -> Iterable["_models.ProviderListResult"]
286        """Gets all resource providers for a subscription.
287
288        :param top: The number of results to return. If null is passed returns all deployments.
289        :type top: int
290        :param expand: The properties to include in the results. For example, use &$expand=metadata in
291         the query string to retrieve resource provider metadata. To include property aliases in
292         response, use $expand=resourceTypes/aliases.
293        :type expand: str
294        :keyword callable cls: A custom type or function that will be passed the direct response
295        :return: An iterator like instance of either ProviderListResult or the result of cls(response)
296        :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.resource.resources.v2021_04_01.models.ProviderListResult]
297        :raises: ~azure.core.exceptions.HttpResponseError
298        """
299        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ProviderListResult"]
300        error_map = {
301            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
302        }
303        error_map.update(kwargs.pop('error_map', {}))
304        api_version = "2021-04-01"
305        accept = "application/json"
306
307        def prepare_request(next_link=None):
308            # Construct headers
309            header_parameters = {}  # type: Dict[str, Any]
310            header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
311
312            if not next_link:
313                # Construct URL
314                url = self.list.metadata['url']  # type: ignore
315                path_format_arguments = {
316                    'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
317                }
318                url = self._client.format_url(url, **path_format_arguments)
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        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, iter(list_of_elem)
340
341        def get_next(next_link=None):
342            request = prepare_request(next_link)
343
344            pipeline_response = 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 ItemPaged(
354            get_next, extract_data
355        )
356    list.metadata = {'url': '/subscriptions/{subscriptionId}/providers'}  # type: ignore
357
358    def list_at_tenant_scope(
359        self,
360        top=None,  # type: Optional[int]
361        expand=None,  # type: Optional[str]
362        **kwargs  # type: Any
363    ):
364        # type: (...) -> Iterable["_models.ProviderListResult"]
365        """Gets all resource providers for the tenant.
366
367        :param top: The number of results to return. If null is passed returns all providers.
368        :type top: int
369        :param expand: The properties to include in the results. For example, use &$expand=metadata in
370         the query string to retrieve resource provider metadata. To include property aliases in
371         response, use $expand=resourceTypes/aliases.
372        :type expand: str
373        :keyword callable cls: A custom type or function that will be passed the direct response
374        :return: An iterator like instance of either ProviderListResult or the result of cls(response)
375        :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.resource.resources.v2021_04_01.models.ProviderListResult]
376        :raises: ~azure.core.exceptions.HttpResponseError
377        """
378        cls = kwargs.pop('cls', None)  # type: ClsType["_models.ProviderListResult"]
379        error_map = {
380            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
381        }
382        error_map.update(kwargs.pop('error_map', {}))
383        api_version = "2021-04-01"
384        accept = "application/json"
385
386        def prepare_request(next_link=None):
387            # Construct headers
388            header_parameters = {}  # type: Dict[str, Any]
389            header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
390
391            if not next_link:
392                # Construct URL
393                url = self.list_at_tenant_scope.metadata['url']  # type: ignore
394                # Construct parameters
395                query_parameters = {}  # type: Dict[str, Any]
396                if top is not None:
397                    query_parameters['$top'] = self._serialize.query("top", top, 'int')
398                if expand is not None:
399                    query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
400                query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
401
402                request = self._client.get(url, query_parameters, header_parameters)
403            else:
404                url = next_link
405                query_parameters = {}  # type: Dict[str, Any]
406                request = self._client.get(url, query_parameters, header_parameters)
407            return request
408
409        def extract_data(pipeline_response):
410            deserialized = self._deserialize('ProviderListResult', pipeline_response)
411            list_of_elem = deserialized.value
412            if cls:
413                list_of_elem = cls(list_of_elem)
414            return deserialized.next_link or None, iter(list_of_elem)
415
416        def get_next(next_link=None):
417            request = prepare_request(next_link)
418
419            pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
420            response = pipeline_response.http_response
421
422            if response.status_code not in [200]:
423                map_error(status_code=response.status_code, response=response, error_map=error_map)
424                raise HttpResponseError(response=response, error_format=ARMErrorFormat)
425
426            return pipeline_response
427
428        return ItemPaged(
429            get_next, extract_data
430        )
431    list_at_tenant_scope.metadata = {'url': '/providers'}  # type: ignore
432
433    def get(
434        self,
435        resource_provider_namespace,  # type: str
436        expand=None,  # type: Optional[str]
437        **kwargs  # type: Any
438    ):
439        # type: (...) -> "_models.Provider"
440        """Gets the specified resource provider.
441
442        :param resource_provider_namespace: The namespace of the resource provider.
443        :type resource_provider_namespace: str
444        :param expand: The $expand query parameter. For example, to include property aliases in
445         response, use $expand=resourceTypes/aliases.
446        :type expand: str
447        :keyword callable cls: A custom type or function that will be passed the direct response
448        :return: Provider, or the result of cls(response)
449        :rtype: ~azure.mgmt.resource.resources.v2021_04_01.models.Provider
450        :raises: ~azure.core.exceptions.HttpResponseError
451        """
452        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Provider"]
453        error_map = {
454            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
455        }
456        error_map.update(kwargs.pop('error_map', {}))
457        api_version = "2021-04-01"
458        accept = "application/json"
459
460        # Construct URL
461        url = self.get.metadata['url']  # type: ignore
462        path_format_arguments = {
463            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
464            'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'),
465        }
466        url = self._client.format_url(url, **path_format_arguments)
467
468        # Construct parameters
469        query_parameters = {}  # type: Dict[str, Any]
470        if expand is not None:
471            query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
472        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
473
474        # Construct headers
475        header_parameters = {}  # type: Dict[str, Any]
476        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
477
478        request = self._client.get(url, query_parameters, header_parameters)
479        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
480        response = pipeline_response.http_response
481
482        if response.status_code not in [200]:
483            map_error(status_code=response.status_code, response=response, error_map=error_map)
484            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
485
486        deserialized = self._deserialize('Provider', pipeline_response)
487
488        if cls:
489            return cls(pipeline_response, deserialized, {})
490
491        return deserialized
492    get.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}'}  # type: ignore
493
494    def get_at_tenant_scope(
495        self,
496        resource_provider_namespace,  # type: str
497        expand=None,  # type: Optional[str]
498        **kwargs  # type: Any
499    ):
500        # type: (...) -> "_models.Provider"
501        """Gets the specified resource provider at the tenant level.
502
503        :param resource_provider_namespace: The namespace of the resource provider.
504        :type resource_provider_namespace: str
505        :param expand: The $expand query parameter. For example, to include property aliases in
506         response, use $expand=resourceTypes/aliases.
507        :type expand: str
508        :keyword callable cls: A custom type or function that will be passed the direct response
509        :return: Provider, or the result of cls(response)
510        :rtype: ~azure.mgmt.resource.resources.v2021_04_01.models.Provider
511        :raises: ~azure.core.exceptions.HttpResponseError
512        """
513        cls = kwargs.pop('cls', None)  # type: ClsType["_models.Provider"]
514        error_map = {
515            401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError
516        }
517        error_map.update(kwargs.pop('error_map', {}))
518        api_version = "2021-04-01"
519        accept = "application/json"
520
521        # Construct URL
522        url = self.get_at_tenant_scope.metadata['url']  # type: ignore
523        path_format_arguments = {
524            'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'),
525        }
526        url = self._client.format_url(url, **path_format_arguments)
527
528        # Construct parameters
529        query_parameters = {}  # type: Dict[str, Any]
530        if expand is not None:
531            query_parameters['$expand'] = self._serialize.query("expand", expand, 'str')
532        query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str')
533
534        # Construct headers
535        header_parameters = {}  # type: Dict[str, Any]
536        header_parameters['Accept'] = self._serialize.header("accept", accept, 'str')
537
538        request = self._client.get(url, query_parameters, header_parameters)
539        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
540        response = pipeline_response.http_response
541
542        if response.status_code not in [200]:
543            map_error(status_code=response.status_code, response=response, error_map=error_map)
544            raise HttpResponseError(response=response, error_format=ARMErrorFormat)
545
546        deserialized = self._deserialize('Provider', pipeline_response)
547
548        if cls:
549            return cls(pipeline_response, deserialized, {})
550
551        return deserialized
552    get_at_tenant_scope.metadata = {'url': '/providers/{resourceProviderNamespace}'}  # type: ignore
553