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 PolicyAssignmentsOperations: 23 """PolicyAssignmentsOperations 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.policy.v2016_04_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 delete( 45 self, 46 scope: str, 47 policy_assignment_name: str, 48 **kwargs: Any 49 ) -> "_models.PolicyAssignment": 50 """Deletes a policy assignment. 51 52 :param scope: The scope of the policy assignment. 53 :type scope: str 54 :param policy_assignment_name: The name of the policy assignment to delete. 55 :type policy_assignment_name: str 56 :keyword callable cls: A custom type or function that will be passed the direct response 57 :return: PolicyAssignment, or the result of cls(response) 58 :rtype: ~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignment 59 :raises: ~azure.core.exceptions.HttpResponseError 60 """ 61 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 62 error_map = { 63 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 64 } 65 error_map.update(kwargs.pop('error_map', {})) 66 api_version = "2016-04-01" 67 accept = "application/json, text/json" 68 69 # Construct URL 70 url = self.delete.metadata['url'] # type: ignore 71 path_format_arguments = { 72 'scope': self._serialize.url("scope", scope, 'str', skip_quote=True), 73 'policyAssignmentName': self._serialize.url("policy_assignment_name", policy_assignment_name, 'str'), 74 } 75 url = self._client.format_url(url, **path_format_arguments) 76 77 # Construct parameters 78 query_parameters = {} # type: Dict[str, Any] 79 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 80 81 # Construct headers 82 header_parameters = {} # type: Dict[str, Any] 83 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 84 85 request = self._client.delete(url, query_parameters, header_parameters) 86 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 87 response = pipeline_response.http_response 88 89 if response.status_code not in [200]: 90 map_error(status_code=response.status_code, response=response, error_map=error_map) 91 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 92 93 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 94 95 if cls: 96 return cls(pipeline_response, deserialized, {}) 97 98 return deserialized 99 delete.metadata = {'url': '/{scope}/providers/Microsoft.Authorization/policyassignments/{policyAssignmentName}'} # type: ignore 100 101 async def create( 102 self, 103 scope: str, 104 policy_assignment_name: str, 105 parameters: "_models.PolicyAssignment", 106 **kwargs: Any 107 ) -> "_models.PolicyAssignment": 108 """Creates a policy assignment. 109 110 Policy assignments are inherited by child resources. For example, when you apply a policy to a 111 resource group that policy is assigned to all resources in the group. 112 113 :param scope: The scope of the policy assignment. 114 :type scope: str 115 :param policy_assignment_name: The name of the policy assignment. 116 :type policy_assignment_name: str 117 :param parameters: Parameters for the policy assignment. 118 :type parameters: ~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignment 119 :keyword callable cls: A custom type or function that will be passed the direct response 120 :return: PolicyAssignment, or the result of cls(response) 121 :rtype: ~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignment 122 :raises: ~azure.core.exceptions.HttpResponseError 123 """ 124 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 125 error_map = { 126 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 127 } 128 error_map.update(kwargs.pop('error_map', {})) 129 api_version = "2016-04-01" 130 content_type = kwargs.pop("content_type", "application/json") 131 accept = "application/json, text/json" 132 133 # Construct URL 134 url = self.create.metadata['url'] # type: ignore 135 path_format_arguments = { 136 'scope': self._serialize.url("scope", scope, 'str', skip_quote=True), 137 'policyAssignmentName': self._serialize.url("policy_assignment_name", policy_assignment_name, 'str'), 138 } 139 url = self._client.format_url(url, **path_format_arguments) 140 141 # Construct parameters 142 query_parameters = {} # type: Dict[str, Any] 143 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 144 145 # Construct headers 146 header_parameters = {} # type: Dict[str, Any] 147 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 148 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 149 150 body_content_kwargs = {} # type: Dict[str, Any] 151 body_content = self._serialize.body(parameters, 'PolicyAssignment') 152 body_content_kwargs['content'] = body_content 153 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 154 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 155 response = pipeline_response.http_response 156 157 if response.status_code not in [201]: 158 map_error(status_code=response.status_code, response=response, error_map=error_map) 159 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 160 161 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 162 163 if cls: 164 return cls(pipeline_response, deserialized, {}) 165 166 return deserialized 167 create.metadata = {'url': '/{scope}/providers/Microsoft.Authorization/policyassignments/{policyAssignmentName}'} # type: ignore 168 169 async def get( 170 self, 171 scope: str, 172 policy_assignment_name: str, 173 **kwargs: Any 174 ) -> "_models.PolicyAssignment": 175 """Gets a policy assignment. 176 177 :param scope: The scope of the policy assignment. 178 :type scope: str 179 :param policy_assignment_name: The name of the policy assignment to get. 180 :type policy_assignment_name: str 181 :keyword callable cls: A custom type or function that will be passed the direct response 182 :return: PolicyAssignment, or the result of cls(response) 183 :rtype: ~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignment 184 :raises: ~azure.core.exceptions.HttpResponseError 185 """ 186 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 187 error_map = { 188 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 189 } 190 error_map.update(kwargs.pop('error_map', {})) 191 api_version = "2016-04-01" 192 accept = "application/json, text/json" 193 194 # Construct URL 195 url = self.get.metadata['url'] # type: ignore 196 path_format_arguments = { 197 'scope': self._serialize.url("scope", scope, 'str', skip_quote=True), 198 'policyAssignmentName': self._serialize.url("policy_assignment_name", policy_assignment_name, 'str'), 199 } 200 url = self._client.format_url(url, **path_format_arguments) 201 202 # Construct parameters 203 query_parameters = {} # type: Dict[str, Any] 204 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 205 206 # Construct headers 207 header_parameters = {} # type: Dict[str, Any] 208 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 209 210 request = self._client.get(url, query_parameters, header_parameters) 211 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 212 response = pipeline_response.http_response 213 214 if response.status_code not in [200]: 215 map_error(status_code=response.status_code, response=response, error_map=error_map) 216 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 217 218 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 219 220 if cls: 221 return cls(pipeline_response, deserialized, {}) 222 223 return deserialized 224 get.metadata = {'url': '/{scope}/providers/Microsoft.Authorization/policyassignments/{policyAssignmentName}'} # type: ignore 225 226 def list_for_resource_group( 227 self, 228 resource_group_name: str, 229 filter: Optional[str] = None, 230 **kwargs: Any 231 ) -> AsyncIterable["_models.PolicyAssignmentListResult"]: 232 """Gets policy assignments for the resource group. 233 234 :param resource_group_name: The name of the resource group that contains policy assignments. 235 :type resource_group_name: str 236 :param filter: The filter to apply on the operation. 237 :type filter: str 238 :keyword callable cls: A custom type or function that will be passed the direct response 239 :return: An iterator like instance of either PolicyAssignmentListResult or the result of cls(response) 240 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignmentListResult] 241 :raises: ~azure.core.exceptions.HttpResponseError 242 """ 243 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignmentListResult"] 244 error_map = { 245 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 246 } 247 error_map.update(kwargs.pop('error_map', {})) 248 api_version = "2016-04-01" 249 accept = "application/json, text/json" 250 251 def prepare_request(next_link=None): 252 # Construct headers 253 header_parameters = {} # type: Dict[str, Any] 254 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 255 256 if not next_link: 257 # Construct URL 258 url = self.list_for_resource_group.metadata['url'] # type: ignore 259 path_format_arguments = { 260 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 261 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 262 } 263 url = self._client.format_url(url, **path_format_arguments) 264 # Construct parameters 265 query_parameters = {} # type: Dict[str, Any] 266 if filter is not None: 267 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str', skip_quote=True) 268 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 269 270 request = self._client.get(url, query_parameters, header_parameters) 271 else: 272 url = next_link 273 query_parameters = {} # type: Dict[str, Any] 274 request = self._client.get(url, query_parameters, header_parameters) 275 return request 276 277 async def extract_data(pipeline_response): 278 deserialized = self._deserialize('PolicyAssignmentListResult', pipeline_response) 279 list_of_elem = deserialized.value 280 if cls: 281 list_of_elem = cls(list_of_elem) 282 return deserialized.next_link or None, AsyncList(list_of_elem) 283 284 async def get_next(next_link=None): 285 request = prepare_request(next_link) 286 287 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 288 response = pipeline_response.http_response 289 290 if response.status_code not in [200]: 291 map_error(status_code=response.status_code, response=response, error_map=error_map) 292 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 293 294 return pipeline_response 295 296 return AsyncItemPaged( 297 get_next, extract_data 298 ) 299 list_for_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Authorization/policyAssignments'} # type: ignore 300 301 def list_for_resource( 302 self, 303 resource_group_name: str, 304 resource_provider_namespace: str, 305 parent_resource_path: str, 306 resource_type: str, 307 resource_name: str, 308 filter: Optional[str] = None, 309 **kwargs: Any 310 ) -> AsyncIterable["_models.PolicyAssignmentListResult"]: 311 """Gets policy assignments for a resource. 312 313 :param resource_group_name: The name of the resource group containing the resource. The name is 314 case insensitive. 315 :type resource_group_name: str 316 :param resource_provider_namespace: The namespace of the resource provider. 317 :type resource_provider_namespace: str 318 :param parent_resource_path: The parent resource path. 319 :type parent_resource_path: str 320 :param resource_type: The resource type. 321 :type resource_type: str 322 :param resource_name: The name of the resource with policy assignments. 323 :type resource_name: str 324 :param filter: The filter to apply on the operation. 325 :type filter: str 326 :keyword callable cls: A custom type or function that will be passed the direct response 327 :return: An iterator like instance of either PolicyAssignmentListResult or the result of cls(response) 328 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignmentListResult] 329 :raises: ~azure.core.exceptions.HttpResponseError 330 """ 331 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignmentListResult"] 332 error_map = { 333 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 334 } 335 error_map.update(kwargs.pop('error_map', {})) 336 api_version = "2016-04-01" 337 accept = "application/json, text/json" 338 339 def prepare_request(next_link=None): 340 # Construct headers 341 header_parameters = {} # type: Dict[str, Any] 342 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 343 344 if not next_link: 345 # Construct URL 346 url = self.list_for_resource.metadata['url'] # type: ignore 347 path_format_arguments = { 348 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 349 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 350 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 351 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 352 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 353 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 354 } 355 url = self._client.format_url(url, **path_format_arguments) 356 # Construct parameters 357 query_parameters = {} # type: Dict[str, Any] 358 if filter is not None: 359 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str') 360 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 361 362 request = self._client.get(url, query_parameters, header_parameters) 363 else: 364 url = next_link 365 query_parameters = {} # type: Dict[str, Any] 366 request = self._client.get(url, query_parameters, header_parameters) 367 return request 368 369 async def extract_data(pipeline_response): 370 deserialized = self._deserialize('PolicyAssignmentListResult', pipeline_response) 371 list_of_elem = deserialized.value 372 if cls: 373 list_of_elem = cls(list_of_elem) 374 return deserialized.next_link or None, AsyncList(list_of_elem) 375 376 async def get_next(next_link=None): 377 request = prepare_request(next_link) 378 379 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 380 response = pipeline_response.http_response 381 382 if response.status_code not in [200]: 383 map_error(status_code=response.status_code, response=response, error_map=error_map) 384 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 385 386 return pipeline_response 387 388 return AsyncItemPaged( 389 get_next, extract_data 390 ) 391 list_for_resource.metadata = {'url': '/subscriptions/{subscriptionId}/resourcegroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}/providers/Microsoft.Authorization/policyassignments'} # type: ignore 392 393 def list( 394 self, 395 filter: Optional[str] = None, 396 **kwargs: Any 397 ) -> AsyncIterable["_models.PolicyAssignmentListResult"]: 398 """Gets all the policy assignments for a subscription. 399 400 :param filter: The filter to apply on the operation. 401 :type filter: str 402 :keyword callable cls: A custom type or function that will be passed the direct response 403 :return: An iterator like instance of either PolicyAssignmentListResult or the result of cls(response) 404 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignmentListResult] 405 :raises: ~azure.core.exceptions.HttpResponseError 406 """ 407 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignmentListResult"] 408 error_map = { 409 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 410 } 411 error_map.update(kwargs.pop('error_map', {})) 412 api_version = "2016-04-01" 413 accept = "application/json, text/json" 414 415 def prepare_request(next_link=None): 416 # Construct headers 417 header_parameters = {} # type: Dict[str, Any] 418 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 419 420 if not next_link: 421 # Construct URL 422 url = self.list.metadata['url'] # type: ignore 423 path_format_arguments = { 424 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 425 } 426 url = self._client.format_url(url, **path_format_arguments) 427 # Construct parameters 428 query_parameters = {} # type: Dict[str, Any] 429 if filter is not None: 430 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str') 431 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 432 433 request = self._client.get(url, query_parameters, header_parameters) 434 else: 435 url = next_link 436 query_parameters = {} # type: Dict[str, Any] 437 request = self._client.get(url, query_parameters, header_parameters) 438 return request 439 440 async def extract_data(pipeline_response): 441 deserialized = self._deserialize('PolicyAssignmentListResult', pipeline_response) 442 list_of_elem = deserialized.value 443 if cls: 444 list_of_elem = cls(list_of_elem) 445 return deserialized.next_link or None, AsyncList(list_of_elem) 446 447 async def get_next(next_link=None): 448 request = prepare_request(next_link) 449 450 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 451 response = pipeline_response.http_response 452 453 if response.status_code not in [200]: 454 map_error(status_code=response.status_code, response=response, error_map=error_map) 455 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 456 457 return pipeline_response 458 459 return AsyncItemPaged( 460 get_next, extract_data 461 ) 462 list.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Authorization/policyassignments'} # type: ignore 463 464 async def delete_by_id( 465 self, 466 policy_assignment_id: str, 467 **kwargs: Any 468 ) -> "_models.PolicyAssignment": 469 """Deletes a policy assignment by ID. 470 471 When providing a scope for the assignment, use '/subscriptions/{subscription-id}/' for 472 subscriptions, '/subscriptions/{subscription-id}/resourceGroups/{resource-group-name}' for 473 resource groups, and 474 '/subscriptions/{subscription-id}/resourceGroups/{resource-group-name}/providers/{resource-provider-namespace}/{resource-type}/{resource-name}' 475 for resources. 476 477 :param policy_assignment_id: The ID of the policy assignment to delete. Use the format 478 '/{scope}/providers/Microsoft.Authorization/policyAssignments/{policy-assignment-name}'. 479 :type policy_assignment_id: str 480 :keyword callable cls: A custom type or function that will be passed the direct response 481 :return: PolicyAssignment, or the result of cls(response) 482 :rtype: ~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignment 483 :raises: ~azure.core.exceptions.HttpResponseError 484 """ 485 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 486 error_map = { 487 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 488 } 489 error_map.update(kwargs.pop('error_map', {})) 490 api_version = "2016-04-01" 491 accept = "application/json, text/json" 492 493 # Construct URL 494 url = self.delete_by_id.metadata['url'] # type: ignore 495 path_format_arguments = { 496 'policyAssignmentId': self._serialize.url("policy_assignment_id", policy_assignment_id, 'str', skip_quote=True), 497 } 498 url = self._client.format_url(url, **path_format_arguments) 499 500 # Construct parameters 501 query_parameters = {} # type: Dict[str, Any] 502 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 503 504 # Construct headers 505 header_parameters = {} # type: Dict[str, Any] 506 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 507 508 request = self._client.delete(url, query_parameters, header_parameters) 509 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 510 response = pipeline_response.http_response 511 512 if response.status_code not in [200]: 513 map_error(status_code=response.status_code, response=response, error_map=error_map) 514 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 515 516 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 517 518 if cls: 519 return cls(pipeline_response, deserialized, {}) 520 521 return deserialized 522 delete_by_id.metadata = {'url': '/{policyAssignmentId}'} # type: ignore 523 524 async def create_by_id( 525 self, 526 policy_assignment_id: str, 527 parameters: "_models.PolicyAssignment", 528 **kwargs: Any 529 ) -> "_models.PolicyAssignment": 530 """Creates a policy assignment by ID. 531 532 Policy assignments are inherited by child resources. For example, when you apply a policy to a 533 resource group that policy is assigned to all resources in the group. When providing a scope 534 for the assignment, use '/subscriptions/{subscription-id}/' for subscriptions, 535 '/subscriptions/{subscription-id}/resourceGroups/{resource-group-name}' for resource groups, 536 and 537 '/subscriptions/{subscription-id}/resourceGroups/{resource-group-name}/providers/{resource-provider-namespace}/{resource-type}/{resource-name}' 538 for resources. 539 540 :param policy_assignment_id: The ID of the policy assignment to create. Use the format 541 '/{scope}/providers/Microsoft.Authorization/policyAssignments/{policy-assignment-name}'. 542 :type policy_assignment_id: str 543 :param parameters: Parameters for policy assignment. 544 :type parameters: ~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignment 545 :keyword callable cls: A custom type or function that will be passed the direct response 546 :return: PolicyAssignment, or the result of cls(response) 547 :rtype: ~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignment 548 :raises: ~azure.core.exceptions.HttpResponseError 549 """ 550 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 551 error_map = { 552 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 553 } 554 error_map.update(kwargs.pop('error_map', {})) 555 api_version = "2016-04-01" 556 content_type = kwargs.pop("content_type", "application/json") 557 accept = "application/json, text/json" 558 559 # Construct URL 560 url = self.create_by_id.metadata['url'] # type: ignore 561 path_format_arguments = { 562 'policyAssignmentId': self._serialize.url("policy_assignment_id", policy_assignment_id, 'str', skip_quote=True), 563 } 564 url = self._client.format_url(url, **path_format_arguments) 565 566 # Construct parameters 567 query_parameters = {} # type: Dict[str, Any] 568 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 569 570 # Construct headers 571 header_parameters = {} # type: Dict[str, Any] 572 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 573 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 574 575 body_content_kwargs = {} # type: Dict[str, Any] 576 body_content = self._serialize.body(parameters, 'PolicyAssignment') 577 body_content_kwargs['content'] = body_content 578 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 579 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 580 response = pipeline_response.http_response 581 582 if response.status_code not in [201]: 583 map_error(status_code=response.status_code, response=response, error_map=error_map) 584 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 585 586 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 587 588 if cls: 589 return cls(pipeline_response, deserialized, {}) 590 591 return deserialized 592 create_by_id.metadata = {'url': '/{policyAssignmentId}'} # type: ignore 593 594 async def get_by_id( 595 self, 596 policy_assignment_id: str, 597 **kwargs: Any 598 ) -> "_models.PolicyAssignment": 599 """Gets a policy assignment by ID. 600 601 When providing a scope for the assignment, use '/subscriptions/{subscription-id}/' for 602 subscriptions, '/subscriptions/{subscription-id}/resourceGroups/{resource-group-name}' for 603 resource groups, and 604 '/subscriptions/{subscription-id}/resourceGroups/{resource-group-name}/providers/{resource-provider-namespace}/{resource-type}/{resource-name}' 605 for resources. 606 607 :param policy_assignment_id: The ID of the policy assignment to get. Use the format 608 '/{scope}/providers/Microsoft.Authorization/policyAssignments/{policy-assignment-name}'. 609 :type policy_assignment_id: str 610 :keyword callable cls: A custom type or function that will be passed the direct response 611 :return: PolicyAssignment, or the result of cls(response) 612 :rtype: ~azure.mgmt.resource.policy.v2016_04_01.models.PolicyAssignment 613 :raises: ~azure.core.exceptions.HttpResponseError 614 """ 615 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 616 error_map = { 617 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 618 } 619 error_map.update(kwargs.pop('error_map', {})) 620 api_version = "2016-04-01" 621 accept = "application/json, text/json" 622 623 # Construct URL 624 url = self.get_by_id.metadata['url'] # type: ignore 625 path_format_arguments = { 626 'policyAssignmentId': self._serialize.url("policy_assignment_id", policy_assignment_id, 'str', skip_quote=True), 627 } 628 url = self._client.format_url(url, **path_format_arguments) 629 630 # Construct parameters 631 query_parameters = {} # type: Dict[str, Any] 632 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 633 634 # Construct headers 635 header_parameters = {} # type: Dict[str, Any] 636 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 637 638 request = self._client.get(url, query_parameters, header_parameters) 639 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 640 response = pipeline_response.http_response 641 642 if response.status_code not in [200]: 643 map_error(status_code=response.status_code, response=response, error_map=error_map) 644 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 645 646 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 647 648 if cls: 649 return cls(pipeline_response, deserialized, {}) 650 651 return deserialized 652 get_by_id.metadata = {'url': '/{policyAssignmentId}'} # type: ignore 653