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