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