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