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