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.v2020_09_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 ) -> Optional["_models.PolicyAssignment"]: 50 """Deletes a policy assignment. 51 52 This operation deletes a policy assignment, given its name and the scope it was created in. The 53 scope of a policy assignment is the part of its ID preceding 54 '/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'. 55 56 :param scope: The scope of the policy assignment. Valid scopes are: management group (format: 57 '/providers/Microsoft.Management/managementGroups/{managementGroup}'), subscription (format: 58 '/subscriptions/{subscriptionId}'), resource group (format: 59 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}', or resource (format: 60 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/{resourceProviderNamespace}/[{parentResourcePath}/]{resourceType}/{resourceName}'. 61 :type scope: str 62 :param policy_assignment_name: The name of the policy assignment to delete. 63 :type policy_assignment_name: str 64 :keyword callable cls: A custom type or function that will be passed the direct response 65 :return: PolicyAssignment, or the result of cls(response) 66 :rtype: ~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignment or None 67 :raises: ~azure.core.exceptions.HttpResponseError 68 """ 69 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.PolicyAssignment"]] 70 error_map = { 71 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 72 } 73 error_map.update(kwargs.pop('error_map', {})) 74 api_version = "2020-09-01" 75 accept = "application/json" 76 77 # Construct URL 78 url = self.delete.metadata['url'] # type: ignore 79 path_format_arguments = { 80 'scope': self._serialize.url("scope", scope, 'str', skip_quote=True), 81 'policyAssignmentName': self._serialize.url("policy_assignment_name", policy_assignment_name, 'str'), 82 } 83 url = self._client.format_url(url, **path_format_arguments) 84 85 # Construct parameters 86 query_parameters = {} # type: Dict[str, Any] 87 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 88 89 # Construct headers 90 header_parameters = {} # type: Dict[str, Any] 91 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 92 93 request = self._client.delete(url, query_parameters, header_parameters) 94 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 95 response = pipeline_response.http_response 96 97 if response.status_code not in [200, 204]: 98 map_error(status_code=response.status_code, response=response, error_map=error_map) 99 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 100 101 deserialized = None 102 if response.status_code == 200: 103 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 104 105 if cls: 106 return cls(pipeline_response, deserialized, {}) 107 108 return deserialized 109 delete.metadata = {'url': '/{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'} # type: ignore 110 111 async def create( 112 self, 113 scope: str, 114 policy_assignment_name: str, 115 parameters: "_models.PolicyAssignment", 116 **kwargs: Any 117 ) -> "_models.PolicyAssignment": 118 """Creates or updates a policy assignment. 119 120 This operation creates or updates a policy assignment with the given scope and name. Policy 121 assignments apply to all resources contained within their scope. For example, when you assign a 122 policy at resource group scope, that policy applies to all resources in the group. 123 124 :param scope: The scope of the policy assignment. Valid scopes are: management group (format: 125 '/providers/Microsoft.Management/managementGroups/{managementGroup}'), subscription (format: 126 '/subscriptions/{subscriptionId}'), resource group (format: 127 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}', or resource (format: 128 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/{resourceProviderNamespace}/[{parentResourcePath}/]{resourceType}/{resourceName}'. 129 :type scope: str 130 :param policy_assignment_name: The name of the policy assignment. 131 :type policy_assignment_name: str 132 :param parameters: Parameters for the policy assignment. 133 :type parameters: ~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignment 134 :keyword callable cls: A custom type or function that will be passed the direct response 135 :return: PolicyAssignment, or the result of cls(response) 136 :rtype: ~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignment 137 :raises: ~azure.core.exceptions.HttpResponseError 138 """ 139 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 140 error_map = { 141 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 142 } 143 error_map.update(kwargs.pop('error_map', {})) 144 api_version = "2020-09-01" 145 content_type = kwargs.pop("content_type", "application/json") 146 accept = "application/json" 147 148 # Construct URL 149 url = self.create.metadata['url'] # type: ignore 150 path_format_arguments = { 151 'scope': self._serialize.url("scope", scope, 'str', skip_quote=True), 152 'policyAssignmentName': self._serialize.url("policy_assignment_name", policy_assignment_name, 'str'), 153 } 154 url = self._client.format_url(url, **path_format_arguments) 155 156 # Construct parameters 157 query_parameters = {} # type: Dict[str, Any] 158 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 159 160 # Construct headers 161 header_parameters = {} # type: Dict[str, Any] 162 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 163 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 164 165 body_content_kwargs = {} # type: Dict[str, Any] 166 body_content = self._serialize.body(parameters, 'PolicyAssignment') 167 body_content_kwargs['content'] = body_content 168 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 169 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 170 response = pipeline_response.http_response 171 172 if response.status_code not in [201]: 173 map_error(status_code=response.status_code, response=response, error_map=error_map) 174 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 175 176 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 177 178 if cls: 179 return cls(pipeline_response, deserialized, {}) 180 181 return deserialized 182 create.metadata = {'url': '/{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'} # type: ignore 183 184 async def get( 185 self, 186 scope: str, 187 policy_assignment_name: str, 188 **kwargs: Any 189 ) -> "_models.PolicyAssignment": 190 """Retrieves a policy assignment. 191 192 This operation retrieves a single policy assignment, given its name and the scope it was 193 created at. 194 195 :param scope: The scope of the policy assignment. Valid scopes are: management group (format: 196 '/providers/Microsoft.Management/managementGroups/{managementGroup}'), subscription (format: 197 '/subscriptions/{subscriptionId}'), resource group (format: 198 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}', or resource (format: 199 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/{resourceProviderNamespace}/[{parentResourcePath}/]{resourceType}/{resourceName}'. 200 :type scope: str 201 :param policy_assignment_name: The name of the policy assignment to get. 202 :type policy_assignment_name: str 203 :keyword callable cls: A custom type or function that will be passed the direct response 204 :return: PolicyAssignment, or the result of cls(response) 205 :rtype: ~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignment 206 :raises: ~azure.core.exceptions.HttpResponseError 207 """ 208 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 209 error_map = { 210 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 211 } 212 error_map.update(kwargs.pop('error_map', {})) 213 api_version = "2020-09-01" 214 accept = "application/json" 215 216 # Construct URL 217 url = self.get.metadata['url'] # type: ignore 218 path_format_arguments = { 219 'scope': self._serialize.url("scope", scope, 'str', skip_quote=True), 220 'policyAssignmentName': self._serialize.url("policy_assignment_name", policy_assignment_name, 'str'), 221 } 222 url = self._client.format_url(url, **path_format_arguments) 223 224 # Construct parameters 225 query_parameters = {} # type: Dict[str, Any] 226 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 227 228 # Construct headers 229 header_parameters = {} # type: Dict[str, Any] 230 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 231 232 request = self._client.get(url, query_parameters, header_parameters) 233 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 234 response = pipeline_response.http_response 235 236 if response.status_code not in [200]: 237 map_error(status_code=response.status_code, response=response, error_map=error_map) 238 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 239 240 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 241 242 if cls: 243 return cls(pipeline_response, deserialized, {}) 244 245 return deserialized 246 get.metadata = {'url': '/{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'} # type: ignore 247 248 def list_for_resource_group( 249 self, 250 resource_group_name: str, 251 filter: Optional[str] = None, 252 top: Optional[int] = None, 253 **kwargs: Any 254 ) -> AsyncIterable["_models.PolicyAssignmentListResult"]: 255 """Retrieves all policy assignments that apply to a resource group. 256 257 This operation retrieves the list of all policy assignments associated with the given resource 258 group in the given subscription that match the optional given $filter. Valid values for $filter 259 are: 'atScope()', 'atExactScope()' or 'policyDefinitionId eq '{value}''. If $filter is not 260 provided, the unfiltered list includes all policy assignments associated with the resource 261 group, including those that apply directly or apply from containing scopes, as well as any 262 applied to resources contained within the resource group. If $filter=atScope() is provided, the 263 returned list includes all policy assignments that apply to the resource group, which is 264 everything in the unfiltered list except those applied to resources contained within the 265 resource group. If $filter=atExactScope() is provided, the returned list only includes all 266 policy assignments that at the resource group. If $filter=policyDefinitionId eq '{value}' is 267 provided, the returned list includes all policy assignments of the policy definition whose id 268 is {value} that apply to the resource group. 269 270 :param resource_group_name: The name of the resource group that contains policy assignments. 271 :type resource_group_name: str 272 :param filter: The filter to apply on the operation. Valid values for $filter are: 'atScope()', 273 'atExactScope()' or 'policyDefinitionId eq '{value}''. If $filter is not provided, no filtering 274 is performed. If $filter=atScope() is provided, the returned list only includes all policy 275 assignments that apply to the scope, which is everything in the unfiltered list except those 276 applied to sub scopes contained within the given scope. If $filter=atExactScope() is provided, 277 the returned list only includes all policy assignments that at the given scope. If 278 $filter=policyDefinitionId eq '{value}' is provided, the returned list includes all policy 279 assignments of the policy definition whose id is {value}. 280 :type filter: str 281 :param top: Maximum number of records to return. When the $top filter is not provided, it will 282 return 500 records. 283 :type top: int 284 :keyword callable cls: A custom type or function that will be passed the direct response 285 :return: An iterator like instance of either PolicyAssignmentListResult or the result of cls(response) 286 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignmentListResult] 287 :raises: ~azure.core.exceptions.HttpResponseError 288 """ 289 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignmentListResult"] 290 error_map = { 291 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 292 } 293 error_map.update(kwargs.pop('error_map', {})) 294 api_version = "2020-09-01" 295 accept = "application/json" 296 297 def prepare_request(next_link=None): 298 # Construct headers 299 header_parameters = {} # type: Dict[str, Any] 300 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 301 302 if not next_link: 303 # Construct URL 304 url = self.list_for_resource_group.metadata['url'] # type: ignore 305 path_format_arguments = { 306 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 307 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 308 } 309 url = self._client.format_url(url, **path_format_arguments) 310 # Construct parameters 311 query_parameters = {} # type: Dict[str, Any] 312 if filter is not None: 313 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str', skip_quote=True) 314 if top is not None: 315 query_parameters['$top'] = self._serialize.query("top", top, 'int', maximum=1000, minimum=1) 316 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 317 318 request = self._client.get(url, query_parameters, header_parameters) 319 else: 320 url = next_link 321 query_parameters = {} # type: Dict[str, Any] 322 request = self._client.get(url, query_parameters, header_parameters) 323 return request 324 325 async def extract_data(pipeline_response): 326 deserialized = self._deserialize('PolicyAssignmentListResult', pipeline_response) 327 list_of_elem = deserialized.value 328 if cls: 329 list_of_elem = cls(list_of_elem) 330 return deserialized.next_link or None, AsyncList(list_of_elem) 331 332 async def get_next(next_link=None): 333 request = prepare_request(next_link) 334 335 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 336 response = pipeline_response.http_response 337 338 if response.status_code not in [200]: 339 map_error(status_code=response.status_code, response=response, error_map=error_map) 340 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 341 342 return pipeline_response 343 344 return AsyncItemPaged( 345 get_next, extract_data 346 ) 347 list_for_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Authorization/policyAssignments'} # type: ignore 348 349 def list_for_resource( 350 self, 351 resource_group_name: str, 352 resource_provider_namespace: str, 353 parent_resource_path: str, 354 resource_type: str, 355 resource_name: str, 356 filter: Optional[str] = None, 357 top: Optional[int] = None, 358 **kwargs: Any 359 ) -> AsyncIterable["_models.PolicyAssignmentListResult"]: 360 """Retrieves all policy assignments that apply to a resource. 361 362 This operation retrieves the list of all policy assignments associated with the specified 363 resource in the given resource group and subscription that match the optional given $filter. 364 Valid values for $filter are: 'atScope()', 'atExactScope()' or 'policyDefinitionId eq 365 '{value}''. If $filter is not provided, the unfiltered list includes all policy assignments 366 associated with the resource, including those that apply directly or from all containing 367 scopes, as well as any applied to resources contained within the resource. If $filter=atScope() 368 is provided, the returned list includes all policy assignments that apply to the resource, 369 which is everything in the unfiltered list except those applied to resources contained within 370 the resource. If $filter=atExactScope() is provided, the returned list only includes all policy 371 assignments that at the resource level. If $filter=policyDefinitionId eq '{value}' is provided, 372 the returned list includes all policy assignments of the policy definition whose id is {value} 373 that apply to the resource. Three parameters plus the resource name are used to identify a 374 specific resource. If the resource is not part of a parent resource (the more common case), the 375 parent resource path should not be provided (or provided as ''). For example a web app could be 376 specified as ({resourceProviderNamespace} == 'Microsoft.Web', {parentResourcePath} == '', 377 {resourceType} == 'sites', {resourceName} == 'MyWebApp'). If the resource is part of a parent 378 resource, then all parameters should be provided. For example a virtual machine DNS name could 379 be specified as ({resourceProviderNamespace} == 'Microsoft.Compute', {parentResourcePath} == 380 'virtualMachines/MyVirtualMachine', {resourceType} == 'domainNames', {resourceName} == 381 'MyComputerName'). A convenient alternative to providing the namespace and type name separately 382 is to provide both in the {resourceType} parameter, format: ({resourceProviderNamespace} == '', 383 {parentResourcePath} == '', {resourceType} == 'Microsoft.Web/sites', {resourceName} == 384 'MyWebApp'). 385 386 :param resource_group_name: The name of the resource group containing the resource. 387 :type resource_group_name: str 388 :param resource_provider_namespace: The namespace of the resource provider. For example, the 389 namespace of a virtual machine is Microsoft.Compute (from Microsoft.Compute/virtualMachines). 390 :type resource_provider_namespace: str 391 :param parent_resource_path: The parent resource path. Use empty string if there is none. 392 :type parent_resource_path: str 393 :param resource_type: The resource type name. For example the type name of a web app is 'sites' 394 (from Microsoft.Web/sites). 395 :type resource_type: str 396 :param resource_name: The name of the resource. 397 :type resource_name: str 398 :param filter: The filter to apply on the operation. Valid values for $filter are: 'atScope()', 399 'atExactScope()' or 'policyDefinitionId eq '{value}''. If $filter is not provided, no filtering 400 is performed. If $filter=atScope() is provided, the returned list only includes all policy 401 assignments that apply to the scope, which is everything in the unfiltered list except those 402 applied to sub scopes contained within the given scope. If $filter=atExactScope() is provided, 403 the returned list only includes all policy assignments that at the given scope. If 404 $filter=policyDefinitionId eq '{value}' is provided, the returned list includes all policy 405 assignments of the policy definition whose id is {value}. 406 :type filter: str 407 :param top: Maximum number of records to return. When the $top filter is not provided, it will 408 return 500 records. 409 :type top: int 410 :keyword callable cls: A custom type or function that will be passed the direct response 411 :return: An iterator like instance of either PolicyAssignmentListResult or the result of cls(response) 412 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignmentListResult] 413 :raises: ~azure.core.exceptions.HttpResponseError 414 """ 415 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignmentListResult"] 416 error_map = { 417 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 418 } 419 error_map.update(kwargs.pop('error_map', {})) 420 api_version = "2020-09-01" 421 accept = "application/json" 422 423 def prepare_request(next_link=None): 424 # Construct headers 425 header_parameters = {} # type: Dict[str, Any] 426 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 427 428 if not next_link: 429 # Construct URL 430 url = self.list_for_resource.metadata['url'] # type: ignore 431 path_format_arguments = { 432 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', max_length=90, min_length=1, pattern=r'^[-\w\._\(\)]+$'), 433 'resourceProviderNamespace': self._serialize.url("resource_provider_namespace", resource_provider_namespace, 'str'), 434 'parentResourcePath': self._serialize.url("parent_resource_path", parent_resource_path, 'str', skip_quote=True), 435 'resourceType': self._serialize.url("resource_type", resource_type, 'str', skip_quote=True), 436 'resourceName': self._serialize.url("resource_name", resource_name, 'str'), 437 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 438 } 439 url = self._client.format_url(url, **path_format_arguments) 440 # Construct parameters 441 query_parameters = {} # type: Dict[str, Any] 442 if filter is not None: 443 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str', skip_quote=True) 444 if top is not None: 445 query_parameters['$top'] = self._serialize.query("top", top, 'int', maximum=1000, minimum=1) 446 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 447 448 request = self._client.get(url, query_parameters, header_parameters) 449 else: 450 url = next_link 451 query_parameters = {} # type: Dict[str, Any] 452 request = self._client.get(url, query_parameters, header_parameters) 453 return request 454 455 async def extract_data(pipeline_response): 456 deserialized = self._deserialize('PolicyAssignmentListResult', pipeline_response) 457 list_of_elem = deserialized.value 458 if cls: 459 list_of_elem = cls(list_of_elem) 460 return deserialized.next_link or None, AsyncList(list_of_elem) 461 462 async def get_next(next_link=None): 463 request = prepare_request(next_link) 464 465 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 466 response = pipeline_response.http_response 467 468 if response.status_code not in [200]: 469 map_error(status_code=response.status_code, response=response, error_map=error_map) 470 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 471 472 return pipeline_response 473 474 return AsyncItemPaged( 475 get_next, extract_data 476 ) 477 list_for_resource.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/{resourceProviderNamespace}/{parentResourcePath}/{resourceType}/{resourceName}/providers/Microsoft.Authorization/policyAssignments'} # type: ignore 478 479 def list_for_management_group( 480 self, 481 management_group_id: str, 482 filter: Optional[str] = None, 483 top: Optional[int] = None, 484 **kwargs: Any 485 ) -> AsyncIterable["_models.PolicyAssignmentListResult"]: 486 """Retrieves all policy assignments that apply to a management group. 487 488 This operation retrieves the list of all policy assignments applicable to the management group 489 that match the given $filter. Valid values for $filter are: 'atScope()', 'atExactScope()' or 490 'policyDefinitionId eq '{value}''. If $filter=atScope() is provided, the returned list includes 491 all policy assignments that are assigned to the management group or the management group's 492 ancestors. If $filter=atExactScope() is provided, the returned list only includes all policy 493 assignments that at the management group. If $filter=policyDefinitionId eq '{value}' is 494 provided, the returned list includes all policy assignments of the policy definition whose id 495 is {value} that apply to the management group. 496 497 :param management_group_id: The ID of the management group. 498 :type management_group_id: str 499 :param filter: The filter to apply on the operation. Valid values for $filter are: 'atScope()', 500 'atExactScope()' or 'policyDefinitionId eq '{value}''. If $filter is not provided, no filtering 501 is performed. If $filter=atScope() is provided, the returned list only includes all policy 502 assignments that apply to the scope, which is everything in the unfiltered list except those 503 applied to sub scopes contained within the given scope. If $filter=atExactScope() is provided, 504 the returned list only includes all policy assignments that at the given scope. If 505 $filter=policyDefinitionId eq '{value}' is provided, the returned list includes all policy 506 assignments of the policy definition whose id is {value}. 507 :type filter: str 508 :param top: Maximum number of records to return. When the $top filter is not provided, it will 509 return 500 records. 510 :type top: int 511 :keyword callable cls: A custom type or function that will be passed the direct response 512 :return: An iterator like instance of either PolicyAssignmentListResult or the result of cls(response) 513 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignmentListResult] 514 :raises: ~azure.core.exceptions.HttpResponseError 515 """ 516 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignmentListResult"] 517 error_map = { 518 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 519 } 520 error_map.update(kwargs.pop('error_map', {})) 521 api_version = "2020-09-01" 522 accept = "application/json" 523 524 def prepare_request(next_link=None): 525 # Construct headers 526 header_parameters = {} # type: Dict[str, Any] 527 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 528 529 if not next_link: 530 # Construct URL 531 url = self.list_for_management_group.metadata['url'] # type: ignore 532 path_format_arguments = { 533 'managementGroupId': self._serialize.url("management_group_id", management_group_id, 'str'), 534 } 535 url = self._client.format_url(url, **path_format_arguments) 536 # Construct parameters 537 query_parameters = {} # type: Dict[str, Any] 538 if filter is not None: 539 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str', skip_quote=True) 540 if top is not None: 541 query_parameters['$top'] = self._serialize.query("top", top, 'int', maximum=1000, minimum=1) 542 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 543 544 request = self._client.get(url, query_parameters, header_parameters) 545 else: 546 url = next_link 547 query_parameters = {} # type: Dict[str, Any] 548 request = self._client.get(url, query_parameters, header_parameters) 549 return request 550 551 async def extract_data(pipeline_response): 552 deserialized = self._deserialize('PolicyAssignmentListResult', pipeline_response) 553 list_of_elem = deserialized.value 554 if cls: 555 list_of_elem = cls(list_of_elem) 556 return deserialized.next_link or None, AsyncList(list_of_elem) 557 558 async def get_next(next_link=None): 559 request = prepare_request(next_link) 560 561 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 562 response = pipeline_response.http_response 563 564 if response.status_code not in [200]: 565 map_error(status_code=response.status_code, response=response, error_map=error_map) 566 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 567 568 return pipeline_response 569 570 return AsyncItemPaged( 571 get_next, extract_data 572 ) 573 list_for_management_group.metadata = {'url': '/providers/Microsoft.Management/managementGroups/{managementGroupId}/providers/Microsoft.Authorization/policyAssignments'} # type: ignore 574 575 def list( 576 self, 577 filter: Optional[str] = None, 578 top: Optional[int] = None, 579 **kwargs: Any 580 ) -> AsyncIterable["_models.PolicyAssignmentListResult"]: 581 """Retrieves all policy assignments that apply to a subscription. 582 583 This operation retrieves the list of all policy assignments associated with the given 584 subscription that match the optional given $filter. Valid values for $filter are: 'atScope()', 585 'atExactScope()' or 'policyDefinitionId eq '{value}''. If $filter is not provided, the 586 unfiltered list includes all policy assignments associated with the subscription, including 587 those that apply directly or from management groups that contain the given subscription, as 588 well as any applied to objects contained within the subscription. If $filter=atScope() is 589 provided, the returned list includes all policy assignments that apply to the subscription, 590 which is everything in the unfiltered list except those applied to objects contained within the 591 subscription. If $filter=atExactScope() is provided, the returned list only includes all policy 592 assignments that at the subscription. If $filter=policyDefinitionId eq '{value}' is provided, 593 the returned list includes all policy assignments of the policy definition whose id is {value}. 594 595 :param filter: The filter to apply on the operation. Valid values for $filter are: 'atScope()', 596 'atExactScope()' or 'policyDefinitionId eq '{value}''. If $filter is not provided, no filtering 597 is performed. If $filter=atScope() is provided, the returned list only includes all policy 598 assignments that apply to the scope, which is everything in the unfiltered list except those 599 applied to sub scopes contained within the given scope. If $filter=atExactScope() is provided, 600 the returned list only includes all policy assignments that at the given scope. If 601 $filter=policyDefinitionId eq '{value}' is provided, the returned list includes all policy 602 assignments of the policy definition whose id is {value}. 603 :type filter: str 604 :param top: Maximum number of records to return. When the $top filter is not provided, it will 605 return 500 records. 606 :type top: int 607 :keyword callable cls: A custom type or function that will be passed the direct response 608 :return: An iterator like instance of either PolicyAssignmentListResult or the result of cls(response) 609 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignmentListResult] 610 :raises: ~azure.core.exceptions.HttpResponseError 611 """ 612 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignmentListResult"] 613 error_map = { 614 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 615 } 616 error_map.update(kwargs.pop('error_map', {})) 617 api_version = "2020-09-01" 618 accept = "application/json" 619 620 def prepare_request(next_link=None): 621 # Construct headers 622 header_parameters = {} # type: Dict[str, Any] 623 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 624 625 if not next_link: 626 # Construct URL 627 url = self.list.metadata['url'] # type: ignore 628 path_format_arguments = { 629 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 630 } 631 url = self._client.format_url(url, **path_format_arguments) 632 # Construct parameters 633 query_parameters = {} # type: Dict[str, Any] 634 if filter is not None: 635 query_parameters['$filter'] = self._serialize.query("filter", filter, 'str', skip_quote=True) 636 if top is not None: 637 query_parameters['$top'] = self._serialize.query("top", top, 'int', maximum=1000, minimum=1) 638 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 639 640 request = self._client.get(url, query_parameters, header_parameters) 641 else: 642 url = next_link 643 query_parameters = {} # type: Dict[str, Any] 644 request = self._client.get(url, query_parameters, header_parameters) 645 return request 646 647 async def extract_data(pipeline_response): 648 deserialized = self._deserialize('PolicyAssignmentListResult', pipeline_response) 649 list_of_elem = deserialized.value 650 if cls: 651 list_of_elem = cls(list_of_elem) 652 return deserialized.next_link or None, AsyncList(list_of_elem) 653 654 async def get_next(next_link=None): 655 request = prepare_request(next_link) 656 657 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 658 response = pipeline_response.http_response 659 660 if response.status_code not in [200]: 661 map_error(status_code=response.status_code, response=response, error_map=error_map) 662 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 663 664 return pipeline_response 665 666 return AsyncItemPaged( 667 get_next, extract_data 668 ) 669 list.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Authorization/policyAssignments'} # type: ignore 670 671 async def delete_by_id( 672 self, 673 policy_assignment_id: str, 674 **kwargs: Any 675 ) -> Optional["_models.PolicyAssignment"]: 676 """Deletes a policy assignment. 677 678 This operation deletes the policy with the given ID. Policy assignment IDs have this format: 679 '{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'. Valid 680 formats for {scope} are: '/providers/Microsoft.Management/managementGroups/{managementGroup}' 681 (management group), '/subscriptions/{subscriptionId}' (subscription), 682 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}' (resource group), or 683 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/{resourceProviderNamespace}/[{parentResourcePath}/]{resourceType}/{resourceName}' 684 (resource). 685 686 :param policy_assignment_id: The ID of the policy assignment to delete. Use the format 687 '{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'. 688 :type policy_assignment_id: str 689 :keyword callable cls: A custom type or function that will be passed the direct response 690 :return: PolicyAssignment, or the result of cls(response) 691 :rtype: ~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignment or None 692 :raises: ~azure.core.exceptions.HttpResponseError 693 """ 694 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.PolicyAssignment"]] 695 error_map = { 696 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 697 } 698 error_map.update(kwargs.pop('error_map', {})) 699 api_version = "2020-09-01" 700 accept = "application/json" 701 702 # Construct URL 703 url = self.delete_by_id.metadata['url'] # type: ignore 704 path_format_arguments = { 705 'policyAssignmentId': self._serialize.url("policy_assignment_id", policy_assignment_id, 'str', skip_quote=True), 706 } 707 url = self._client.format_url(url, **path_format_arguments) 708 709 # Construct parameters 710 query_parameters = {} # type: Dict[str, Any] 711 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 712 713 # Construct headers 714 header_parameters = {} # type: Dict[str, Any] 715 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 716 717 request = self._client.delete(url, query_parameters, header_parameters) 718 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 719 response = pipeline_response.http_response 720 721 if response.status_code not in [200, 204]: 722 map_error(status_code=response.status_code, response=response, error_map=error_map) 723 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 724 725 deserialized = None 726 if response.status_code == 200: 727 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 728 729 if cls: 730 return cls(pipeline_response, deserialized, {}) 731 732 return deserialized 733 delete_by_id.metadata = {'url': '/{policyAssignmentId}'} # type: ignore 734 735 async def create_by_id( 736 self, 737 policy_assignment_id: str, 738 parameters: "_models.PolicyAssignment", 739 **kwargs: Any 740 ) -> "_models.PolicyAssignment": 741 """Creates or updates a policy assignment. 742 743 This operation creates or updates the policy assignment with the given ID. Policy assignments 744 made on a scope apply to all resources contained in that scope. For example, when you assign a 745 policy to a resource group that policy applies to all resources in the group. Policy assignment 746 IDs have this format: 747 '{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'. Valid 748 scopes are: management group (format: 749 '/providers/Microsoft.Management/managementGroups/{managementGroup}'), subscription (format: 750 '/subscriptions/{subscriptionId}'), resource group (format: 751 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}', or resource (format: 752 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/{resourceProviderNamespace}/[{parentResourcePath}/]{resourceType}/{resourceName}'. 753 754 :param policy_assignment_id: The ID of the policy assignment to create. Use the format 755 '{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'. 756 :type policy_assignment_id: str 757 :param parameters: Parameters for policy assignment. 758 :type parameters: ~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignment 759 :keyword callable cls: A custom type or function that will be passed the direct response 760 :return: PolicyAssignment, or the result of cls(response) 761 :rtype: ~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignment 762 :raises: ~azure.core.exceptions.HttpResponseError 763 """ 764 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 765 error_map = { 766 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 767 } 768 error_map.update(kwargs.pop('error_map', {})) 769 api_version = "2020-09-01" 770 content_type = kwargs.pop("content_type", "application/json") 771 accept = "application/json" 772 773 # Construct URL 774 url = self.create_by_id.metadata['url'] # type: ignore 775 path_format_arguments = { 776 'policyAssignmentId': self._serialize.url("policy_assignment_id", policy_assignment_id, 'str', skip_quote=True), 777 } 778 url = self._client.format_url(url, **path_format_arguments) 779 780 # Construct parameters 781 query_parameters = {} # type: Dict[str, Any] 782 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 783 784 # Construct headers 785 header_parameters = {} # type: Dict[str, Any] 786 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 787 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 788 789 body_content_kwargs = {} # type: Dict[str, Any] 790 body_content = self._serialize.body(parameters, 'PolicyAssignment') 791 body_content_kwargs['content'] = body_content 792 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 793 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 794 response = pipeline_response.http_response 795 796 if response.status_code not in [201]: 797 map_error(status_code=response.status_code, response=response, error_map=error_map) 798 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 799 800 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 801 802 if cls: 803 return cls(pipeline_response, deserialized, {}) 804 805 return deserialized 806 create_by_id.metadata = {'url': '/{policyAssignmentId}'} # type: ignore 807 808 async def get_by_id( 809 self, 810 policy_assignment_id: str, 811 **kwargs: Any 812 ) -> "_models.PolicyAssignment": 813 """Retrieves the policy assignment with the given ID. 814 815 The operation retrieves the policy assignment with the given ID. Policy assignment IDs have 816 this format: 817 '{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'. Valid 818 scopes are: management group (format: 819 '/providers/Microsoft.Management/managementGroups/{managementGroup}'), subscription (format: 820 '/subscriptions/{subscriptionId}'), resource group (format: 821 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}', or resource (format: 822 '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/{resourceProviderNamespace}/[{parentResourcePath}/]{resourceType}/{resourceName}'. 823 824 :param policy_assignment_id: The ID of the policy assignment to get. Use the format 825 '{scope}/providers/Microsoft.Authorization/policyAssignments/{policyAssignmentName}'. 826 :type policy_assignment_id: str 827 :keyword callable cls: A custom type or function that will be passed the direct response 828 :return: PolicyAssignment, or the result of cls(response) 829 :rtype: ~azure.mgmt.resource.policy.v2020_09_01.models.PolicyAssignment 830 :raises: ~azure.core.exceptions.HttpResponseError 831 """ 832 cls = kwargs.pop('cls', None) # type: ClsType["_models.PolicyAssignment"] 833 error_map = { 834 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 835 } 836 error_map.update(kwargs.pop('error_map', {})) 837 api_version = "2020-09-01" 838 accept = "application/json" 839 840 # Construct URL 841 url = self.get_by_id.metadata['url'] # type: ignore 842 path_format_arguments = { 843 'policyAssignmentId': self._serialize.url("policy_assignment_id", policy_assignment_id, 'str', skip_quote=True), 844 } 845 url = self._client.format_url(url, **path_format_arguments) 846 847 # Construct parameters 848 query_parameters = {} # type: Dict[str, Any] 849 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 850 851 # Construct headers 852 header_parameters = {} # type: Dict[str, Any] 853 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 854 855 request = self._client.get(url, query_parameters, header_parameters) 856 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 857 response = pipeline_response.http_response 858 859 if response.status_code not in [200]: 860 map_error(status_code=response.status_code, response=response, error_map=error_map) 861 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 862 863 deserialized = self._deserialize('PolicyAssignment', pipeline_response) 864 865 if cls: 866 return cls(pipeline_response, deserialized, {}) 867 868 return deserialized 869 get_by_id.metadata = {'url': '/{policyAssignmentId}'} # type: ignore 870