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