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.v2019_10_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.v2019_10_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 = "2019-10-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( 104 self, 105 resource_provider_namespace, # type: str 106 **kwargs # type: Any 107 ): 108 # type: (...) -> "_models.Provider" 109 """Registers a subscription with a resource provider. 110 111 :param resource_provider_namespace: The namespace of the resource provider to register. 112 :type resource_provider_namespace: str 113 :keyword callable cls: A custom type or function that will be passed the direct response 114 :return: Provider, or the result of cls(response) 115 :rtype: ~azure.mgmt.resource.resources.v2019_10_01.models.Provider 116 :raises: ~azure.core.exceptions.HttpResponseError 117 """ 118 cls = kwargs.pop('cls', None) # type: ClsType["_models.Provider"] 119 error_map = { 120 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 121 } 122 error_map.update(kwargs.pop('error_map', {})) 123 api_version = "2019-10-01" 124 accept = "application/json" 125 126 # Construct URL 127 url = self.register.metadata['url'] # type: ignore 128 path_format_arguments = { 129 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 130 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 131 } 132 url = self._client.format_url(url, **path_format_arguments) 133 134 # Construct parameters 135 query_parameters = {} # type: Dict[str, Any] 136 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 137 138 # Construct headers 139 header_parameters = {} # type: Dict[str, Any] 140 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 141 142 request = self._client.post(url, query_parameters, header_parameters) 143 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 144 response = pipeline_response.http_response 145 146 if response.status_code not in [200]: 147 map_error(status_code=response.status_code, response=response, error_map=error_map) 148 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 149 150 deserialized = self._deserialize('Provider', pipeline_response) 151 152 if cls: 153 return cls(pipeline_response, deserialized, {}) 154 155 return deserialized 156 register.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}/register'} # type: ignore 157 158 def list( 159 self, 160 top=None, # type: Optional[int] 161 expand=None, # type: Optional[str] 162 **kwargs # type: Any 163 ): 164 # type: (...) -> Iterable["_models.ProviderListResult"] 165 """Gets all resource providers for a subscription. 166 167 :param top: The number of results to return. If null is passed returns all deployments. 168 :type top: int 169 :param expand: The properties to include in the results. For example, use &$expand=metadata in 170 the query string to retrieve resource provider metadata. To include property aliases in 171 response, use $expand=resourceTypes/aliases. 172 :type expand: str 173 :keyword callable cls: A custom type or function that will be passed the direct response 174 :return: An iterator like instance of either ProviderListResult or the result of cls(response) 175 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.resource.resources.v2019_10_01.models.ProviderListResult] 176 :raises: ~azure.core.exceptions.HttpResponseError 177 """ 178 cls = kwargs.pop('cls', None) # type: ClsType["_models.ProviderListResult"] 179 error_map = { 180 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 181 } 182 error_map.update(kwargs.pop('error_map', {})) 183 api_version = "2019-10-01" 184 accept = "application/json" 185 186 def prepare_request(next_link=None): 187 # Construct headers 188 header_parameters = {} # type: Dict[str, Any] 189 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 190 191 if not next_link: 192 # Construct URL 193 url = self.list.metadata['url'] # type: ignore 194 path_format_arguments = { 195 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 196 } 197 url = self._client.format_url(url, **path_format_arguments) 198 # Construct parameters 199 query_parameters = {} # type: Dict[str, Any] 200 if top is not None: 201 query_parameters['$top'] = self._serialize.query("top", top, 'int') 202 if expand is not None: 203 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 204 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 205 206 request = self._client.get(url, query_parameters, header_parameters) 207 else: 208 url = next_link 209 query_parameters = {} # type: Dict[str, Any] 210 request = self._client.get(url, query_parameters, header_parameters) 211 return request 212 213 def extract_data(pipeline_response): 214 deserialized = self._deserialize('ProviderListResult', pipeline_response) 215 list_of_elem = deserialized.value 216 if cls: 217 list_of_elem = cls(list_of_elem) 218 return deserialized.next_link or None, iter(list_of_elem) 219 220 def get_next(next_link=None): 221 request = prepare_request(next_link) 222 223 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 224 response = pipeline_response.http_response 225 226 if response.status_code not in [200]: 227 map_error(status_code=response.status_code, response=response, error_map=error_map) 228 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 229 230 return pipeline_response 231 232 return ItemPaged( 233 get_next, extract_data 234 ) 235 list.metadata = {'url': '/subscriptions/{subscriptionId}/providers'} # type: ignore 236 237 def list_at_tenant_scope( 238 self, 239 top=None, # type: Optional[int] 240 expand=None, # type: Optional[str] 241 **kwargs # type: Any 242 ): 243 # type: (...) -> Iterable["_models.ProviderListResult"] 244 """Gets all resource providers for the tenant. 245 246 :param top: The number of results to return. If null is passed returns all providers. 247 :type top: int 248 :param expand: The properties to include in the results. For example, use &$expand=metadata in 249 the query string to retrieve resource provider metadata. To include property aliases in 250 response, use $expand=resourceTypes/aliases. 251 :type expand: str 252 :keyword callable cls: A custom type or function that will be passed the direct response 253 :return: An iterator like instance of either ProviderListResult or the result of cls(response) 254 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.resource.resources.v2019_10_01.models.ProviderListResult] 255 :raises: ~azure.core.exceptions.HttpResponseError 256 """ 257 cls = kwargs.pop('cls', None) # type: ClsType["_models.ProviderListResult"] 258 error_map = { 259 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 260 } 261 error_map.update(kwargs.pop('error_map', {})) 262 api_version = "2019-10-01" 263 accept = "application/json" 264 265 def prepare_request(next_link=None): 266 # Construct headers 267 header_parameters = {} # type: Dict[str, Any] 268 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 269 270 if not next_link: 271 # Construct URL 272 url = self.list_at_tenant_scope.metadata['url'] # type: ignore 273 # Construct parameters 274 query_parameters = {} # type: Dict[str, Any] 275 if top is not None: 276 query_parameters['$top'] = self._serialize.query("top", top, 'int') 277 if expand is not None: 278 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 279 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 280 281 request = self._client.get(url, query_parameters, header_parameters) 282 else: 283 url = next_link 284 query_parameters = {} # type: Dict[str, Any] 285 request = self._client.get(url, query_parameters, header_parameters) 286 return request 287 288 def extract_data(pipeline_response): 289 deserialized = self._deserialize('ProviderListResult', pipeline_response) 290 list_of_elem = deserialized.value 291 if cls: 292 list_of_elem = cls(list_of_elem) 293 return deserialized.next_link or None, iter(list_of_elem) 294 295 def get_next(next_link=None): 296 request = prepare_request(next_link) 297 298 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 299 response = pipeline_response.http_response 300 301 if response.status_code not in [200]: 302 map_error(status_code=response.status_code, response=response, error_map=error_map) 303 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 304 305 return pipeline_response 306 307 return ItemPaged( 308 get_next, extract_data 309 ) 310 list_at_tenant_scope.metadata = {'url': '/providers'} # type: ignore 311 312 def get( 313 self, 314 resource_provider_namespace, # type: str 315 expand=None, # type: Optional[str] 316 **kwargs # type: Any 317 ): 318 # type: (...) -> "_models.Provider" 319 """Gets the specified resource provider. 320 321 :param resource_provider_namespace: The namespace of the resource provider. 322 :type resource_provider_namespace: str 323 :param expand: The $expand query parameter. For example, to include property aliases in 324 response, use $expand=resourceTypes/aliases. 325 :type expand: str 326 :keyword callable cls: A custom type or function that will be passed the direct response 327 :return: Provider, or the result of cls(response) 328 :rtype: ~azure.mgmt.resource.resources.v2019_10_01.models.Provider 329 :raises: ~azure.core.exceptions.HttpResponseError 330 """ 331 cls = kwargs.pop('cls', None) # type: ClsType["_models.Provider"] 332 error_map = { 333 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 334 } 335 error_map.update(kwargs.pop('error_map', {})) 336 api_version = "2019-10-01" 337 accept = "application/json" 338 339 # Construct URL 340 url = self.get.metadata['url'] # type: ignore 341 path_format_arguments = { 342 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 343 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 344 } 345 url = self._client.format_url(url, **path_format_arguments) 346 347 # Construct parameters 348 query_parameters = {} # type: Dict[str, Any] 349 if expand is not None: 350 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 351 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 352 353 # Construct headers 354 header_parameters = {} # type: Dict[str, Any] 355 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 356 357 request = self._client.get(url, query_parameters, header_parameters) 358 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 359 response = pipeline_response.http_response 360 361 if response.status_code not in [200]: 362 map_error(status_code=response.status_code, response=response, error_map=error_map) 363 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 364 365 deserialized = self._deserialize('Provider', pipeline_response) 366 367 if cls: 368 return cls(pipeline_response, deserialized, {}) 369 370 return deserialized 371 get.metadata = {'url': '/subscriptions/{subscriptionId}/providers/{resourceProviderNamespace}'} # type: ignore 372 373 def get_at_tenant_scope( 374 self, 375 resource_provider_namespace, # type: str 376 expand=None, # type: Optional[str] 377 **kwargs # type: Any 378 ): 379 # type: (...) -> "_models.Provider" 380 """Gets the specified resource provider at the tenant level. 381 382 :param resource_provider_namespace: The namespace of the resource provider. 383 :type resource_provider_namespace: str 384 :param expand: The $expand query parameter. For example, to include property aliases in 385 response, use $expand=resourceTypes/aliases. 386 :type expand: str 387 :keyword callable cls: A custom type or function that will be passed the direct response 388 :return: Provider, or the result of cls(response) 389 :rtype: ~azure.mgmt.resource.resources.v2019_10_01.models.Provider 390 :raises: ~azure.core.exceptions.HttpResponseError 391 """ 392 cls = kwargs.pop('cls', None) # type: ClsType["_models.Provider"] 393 error_map = { 394 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 395 } 396 error_map.update(kwargs.pop('error_map', {})) 397 api_version = "2019-10-01" 398 accept = "application/json" 399 400 # Construct URL 401 url = self.get_at_tenant_scope.metadata['url'] # type: ignore 402 path_format_arguments = { 403 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 404 } 405 url = self._client.format_url(url, **path_format_arguments) 406 407 # Construct parameters 408 query_parameters = {} # type: Dict[str, Any] 409 if expand is not None: 410 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 411 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 412 413 # Construct headers 414 header_parameters = {} # type: Dict[str, Any] 415 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 416 417 request = self._client.get(url, query_parameters, header_parameters) 418 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 419 response = pipeline_response.http_response 420 421 if response.status_code not in [200]: 422 map_error(status_code=response.status_code, response=response, error_map=error_map) 423 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 424 425 deserialized = self._deserialize('Provider', pipeline_response) 426 427 if cls: 428 return cls(pipeline_response, deserialized, {}) 429 430 return deserialized 431 get_at_tenant_scope.metadata = {'url': '/providers/{resourceProviderNamespace}'} # type: ignore 432