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, Union 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.core.polling import AsyncLROPoller, AsyncNoPolling, AsyncPollingMethod 16from azure.mgmt.core.exceptions import ARMErrorFormat 17from azure.mgmt.core.polling.async_arm_polling import AsyncARMPolling 18 19from ... import models as _models 20 21T = TypeVar('T') 22ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]] 23 24class ServiceEndpointPolicyDefinitionsOperations: 25 """ServiceEndpointPolicyDefinitionsOperations async operations. 26 27 You should not instantiate this class directly. Instead, you should create a Client instance that 28 instantiates it for you and attaches it as an attribute. 29 30 :ivar models: Alias to model classes used in this operation group. 31 :type models: ~azure.mgmt.network.v2020_03_01.models 32 :param client: Client for service requests. 33 :param config: Configuration of service client. 34 :param serializer: An object model serializer. 35 :param deserializer: An object model deserializer. 36 """ 37 38 models = _models 39 40 def __init__(self, client, config, serializer, deserializer) -> None: 41 self._client = client 42 self._serialize = serializer 43 self._deserialize = deserializer 44 self._config = config 45 46 async def _delete_initial( 47 self, 48 resource_group_name: str, 49 service_endpoint_policy_name: str, 50 service_endpoint_policy_definition_name: str, 51 **kwargs 52 ) -> None: 53 cls = kwargs.pop('cls', None) # type: ClsType[None] 54 error_map = { 55 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 56 } 57 error_map.update(kwargs.pop('error_map', {})) 58 api_version = "2020-03-01" 59 accept = "application/json" 60 61 # Construct URL 62 url = self._delete_initial.metadata['url'] # type: ignore 63 path_format_arguments = { 64 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 65 'serviceEndpointPolicyName': self._serialize.url("service_endpoint_policy_name", service_endpoint_policy_name, 'str'), 66 'serviceEndpointPolicyDefinitionName': self._serialize.url("service_endpoint_policy_definition_name", service_endpoint_policy_definition_name, 'str'), 67 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 68 } 69 url = self._client.format_url(url, **path_format_arguments) 70 71 # Construct parameters 72 query_parameters = {} # type: Dict[str, Any] 73 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 74 75 # Construct headers 76 header_parameters = {} # type: Dict[str, Any] 77 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 78 79 request = self._client.delete(url, query_parameters, header_parameters) 80 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 81 response = pipeline_response.http_response 82 83 if response.status_code not in [200, 202, 204]: 84 map_error(status_code=response.status_code, response=response, error_map=error_map) 85 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 86 87 if cls: 88 return cls(pipeline_response, None, {}) 89 90 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/serviceEndpointPolicies/{serviceEndpointPolicyName}/serviceEndpointPolicyDefinitions/{serviceEndpointPolicyDefinitionName}'} # type: ignore 91 92 async def begin_delete( 93 self, 94 resource_group_name: str, 95 service_endpoint_policy_name: str, 96 service_endpoint_policy_definition_name: str, 97 **kwargs 98 ) -> AsyncLROPoller[None]: 99 """Deletes the specified ServiceEndpoint policy definitions. 100 101 :param resource_group_name: The name of the resource group. 102 :type resource_group_name: str 103 :param service_endpoint_policy_name: The name of the Service Endpoint Policy. 104 :type service_endpoint_policy_name: str 105 :param service_endpoint_policy_definition_name: The name of the service endpoint policy 106 definition. 107 :type service_endpoint_policy_definition_name: str 108 :keyword callable cls: A custom type or function that will be passed the direct response 109 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 110 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 111 False for no polling, or your own initialized polling object for a personal polling strategy. 112 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 113 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 114 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 115 :rtype: ~azure.core.polling.AsyncLROPoller[None] 116 :raises ~azure.core.exceptions.HttpResponseError: 117 """ 118 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 119 cls = kwargs.pop('cls', None) # type: ClsType[None] 120 lro_delay = kwargs.pop( 121 'polling_interval', 122 self._config.polling_interval 123 ) 124 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 125 if cont_token is None: 126 raw_result = await self._delete_initial( 127 resource_group_name=resource_group_name, 128 service_endpoint_policy_name=service_endpoint_policy_name, 129 service_endpoint_policy_definition_name=service_endpoint_policy_definition_name, 130 cls=lambda x,y,z: x, 131 **kwargs 132 ) 133 134 kwargs.pop('error_map', None) 135 kwargs.pop('content_type', None) 136 137 def get_long_running_output(pipeline_response): 138 if cls: 139 return cls(pipeline_response, None, {}) 140 141 path_format_arguments = { 142 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 143 'serviceEndpointPolicyName': self._serialize.url("service_endpoint_policy_name", service_endpoint_policy_name, 'str'), 144 'serviceEndpointPolicyDefinitionName': self._serialize.url("service_endpoint_policy_definition_name", service_endpoint_policy_definition_name, 'str'), 145 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 146 } 147 148 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 149 elif polling is False: polling_method = AsyncNoPolling() 150 else: polling_method = polling 151 if cont_token: 152 return AsyncLROPoller.from_continuation_token( 153 polling_method=polling_method, 154 continuation_token=cont_token, 155 client=self._client, 156 deserialization_callback=get_long_running_output 157 ) 158 else: 159 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 160 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/serviceEndpointPolicies/{serviceEndpointPolicyName}/serviceEndpointPolicyDefinitions/{serviceEndpointPolicyDefinitionName}'} # type: ignore 161 162 async def get( 163 self, 164 resource_group_name: str, 165 service_endpoint_policy_name: str, 166 service_endpoint_policy_definition_name: str, 167 **kwargs 168 ) -> "_models.ServiceEndpointPolicyDefinition": 169 """Get the specified service endpoint policy definitions from service endpoint policy. 170 171 :param resource_group_name: The name of the resource group. 172 :type resource_group_name: str 173 :param service_endpoint_policy_name: The name of the service endpoint policy name. 174 :type service_endpoint_policy_name: str 175 :param service_endpoint_policy_definition_name: The name of the service endpoint policy 176 definition name. 177 :type service_endpoint_policy_definition_name: str 178 :keyword callable cls: A custom type or function that will be passed the direct response 179 :return: ServiceEndpointPolicyDefinition, or the result of cls(response) 180 :rtype: ~azure.mgmt.network.v2020_03_01.models.ServiceEndpointPolicyDefinition 181 :raises: ~azure.core.exceptions.HttpResponseError 182 """ 183 cls = kwargs.pop('cls', None) # type: ClsType["_models.ServiceEndpointPolicyDefinition"] 184 error_map = { 185 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 186 } 187 error_map.update(kwargs.pop('error_map', {})) 188 api_version = "2020-03-01" 189 accept = "application/json" 190 191 # Construct URL 192 url = self.get.metadata['url'] # type: ignore 193 path_format_arguments = { 194 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 195 'serviceEndpointPolicyName': self._serialize.url("service_endpoint_policy_name", service_endpoint_policy_name, 'str'), 196 'serviceEndpointPolicyDefinitionName': self._serialize.url("service_endpoint_policy_definition_name", service_endpoint_policy_definition_name, 'str'), 197 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 198 } 199 url = self._client.format_url(url, **path_format_arguments) 200 201 # Construct parameters 202 query_parameters = {} # type: Dict[str, Any] 203 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 204 205 # Construct headers 206 header_parameters = {} # type: Dict[str, Any] 207 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 208 209 request = self._client.get(url, query_parameters, header_parameters) 210 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 211 response = pipeline_response.http_response 212 213 if response.status_code not in [200]: 214 map_error(status_code=response.status_code, response=response, error_map=error_map) 215 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 216 217 deserialized = self._deserialize('ServiceEndpointPolicyDefinition', pipeline_response) 218 219 if cls: 220 return cls(pipeline_response, deserialized, {}) 221 222 return deserialized 223 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/serviceEndpointPolicies/{serviceEndpointPolicyName}/serviceEndpointPolicyDefinitions/{serviceEndpointPolicyDefinitionName}'} # type: ignore 224 225 async def _create_or_update_initial( 226 self, 227 resource_group_name: str, 228 service_endpoint_policy_name: str, 229 service_endpoint_policy_definition_name: str, 230 service_endpoint_policy_definitions: "_models.ServiceEndpointPolicyDefinition", 231 **kwargs 232 ) -> "_models.ServiceEndpointPolicyDefinition": 233 cls = kwargs.pop('cls', None) # type: ClsType["_models.ServiceEndpointPolicyDefinition"] 234 error_map = { 235 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 236 } 237 error_map.update(kwargs.pop('error_map', {})) 238 api_version = "2020-03-01" 239 content_type = kwargs.pop("content_type", "application/json") 240 accept = "application/json" 241 242 # Construct URL 243 url = self._create_or_update_initial.metadata['url'] # type: ignore 244 path_format_arguments = { 245 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 246 'serviceEndpointPolicyName': self._serialize.url("service_endpoint_policy_name", service_endpoint_policy_name, 'str'), 247 'serviceEndpointPolicyDefinitionName': self._serialize.url("service_endpoint_policy_definition_name", service_endpoint_policy_definition_name, 'str'), 248 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 249 } 250 url = self._client.format_url(url, **path_format_arguments) 251 252 # Construct parameters 253 query_parameters = {} # type: Dict[str, Any] 254 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 255 256 # Construct headers 257 header_parameters = {} # type: Dict[str, Any] 258 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 259 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 260 261 body_content_kwargs = {} # type: Dict[str, Any] 262 body_content = self._serialize.body(service_endpoint_policy_definitions, 'ServiceEndpointPolicyDefinition') 263 body_content_kwargs['content'] = body_content 264 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 265 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 266 response = pipeline_response.http_response 267 268 if response.status_code not in [200, 201]: 269 map_error(status_code=response.status_code, response=response, error_map=error_map) 270 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 271 272 if response.status_code == 200: 273 deserialized = self._deserialize('ServiceEndpointPolicyDefinition', pipeline_response) 274 275 if response.status_code == 201: 276 deserialized = self._deserialize('ServiceEndpointPolicyDefinition', pipeline_response) 277 278 if cls: 279 return cls(pipeline_response, deserialized, {}) 280 281 return deserialized 282 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/serviceEndpointPolicies/{serviceEndpointPolicyName}/serviceEndpointPolicyDefinitions/{serviceEndpointPolicyDefinitionName}'} # type: ignore 283 284 async def begin_create_or_update( 285 self, 286 resource_group_name: str, 287 service_endpoint_policy_name: str, 288 service_endpoint_policy_definition_name: str, 289 service_endpoint_policy_definitions: "_models.ServiceEndpointPolicyDefinition", 290 **kwargs 291 ) -> AsyncLROPoller["_models.ServiceEndpointPolicyDefinition"]: 292 """Creates or updates a service endpoint policy definition in the specified service endpoint 293 policy. 294 295 :param resource_group_name: The name of the resource group. 296 :type resource_group_name: str 297 :param service_endpoint_policy_name: The name of the service endpoint policy. 298 :type service_endpoint_policy_name: str 299 :param service_endpoint_policy_definition_name: The name of the service endpoint policy 300 definition name. 301 :type service_endpoint_policy_definition_name: str 302 :param service_endpoint_policy_definitions: Parameters supplied to the create or update service 303 endpoint policy operation. 304 :type service_endpoint_policy_definitions: ~azure.mgmt.network.v2020_03_01.models.ServiceEndpointPolicyDefinition 305 :keyword callable cls: A custom type or function that will be passed the direct response 306 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 307 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 308 False for no polling, or your own initialized polling object for a personal polling strategy. 309 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 310 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 311 :return: An instance of AsyncLROPoller that returns either ServiceEndpointPolicyDefinition or the result of cls(response) 312 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_03_01.models.ServiceEndpointPolicyDefinition] 313 :raises ~azure.core.exceptions.HttpResponseError: 314 """ 315 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 316 cls = kwargs.pop('cls', None) # type: ClsType["_models.ServiceEndpointPolicyDefinition"] 317 lro_delay = kwargs.pop( 318 'polling_interval', 319 self._config.polling_interval 320 ) 321 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 322 if cont_token is None: 323 raw_result = await self._create_or_update_initial( 324 resource_group_name=resource_group_name, 325 service_endpoint_policy_name=service_endpoint_policy_name, 326 service_endpoint_policy_definition_name=service_endpoint_policy_definition_name, 327 service_endpoint_policy_definitions=service_endpoint_policy_definitions, 328 cls=lambda x,y,z: x, 329 **kwargs 330 ) 331 332 kwargs.pop('error_map', None) 333 kwargs.pop('content_type', None) 334 335 def get_long_running_output(pipeline_response): 336 deserialized = self._deserialize('ServiceEndpointPolicyDefinition', pipeline_response) 337 338 if cls: 339 return cls(pipeline_response, deserialized, {}) 340 return deserialized 341 342 path_format_arguments = { 343 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 344 'serviceEndpointPolicyName': self._serialize.url("service_endpoint_policy_name", service_endpoint_policy_name, 'str'), 345 'serviceEndpointPolicyDefinitionName': self._serialize.url("service_endpoint_policy_definition_name", service_endpoint_policy_definition_name, 'str'), 346 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 347 } 348 349 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 350 elif polling is False: polling_method = AsyncNoPolling() 351 else: polling_method = polling 352 if cont_token: 353 return AsyncLROPoller.from_continuation_token( 354 polling_method=polling_method, 355 continuation_token=cont_token, 356 client=self._client, 357 deserialization_callback=get_long_running_output 358 ) 359 else: 360 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 361 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/serviceEndpointPolicies/{serviceEndpointPolicyName}/serviceEndpointPolicyDefinitions/{serviceEndpointPolicyDefinitionName}'} # type: ignore 362 363 def list_by_resource_group( 364 self, 365 resource_group_name: str, 366 service_endpoint_policy_name: str, 367 **kwargs 368 ) -> AsyncIterable["_models.ServiceEndpointPolicyDefinitionListResult"]: 369 """Gets all service endpoint policy definitions in a service end point policy. 370 371 :param resource_group_name: The name of the resource group. 372 :type resource_group_name: str 373 :param service_endpoint_policy_name: The name of the service endpoint policy name. 374 :type service_endpoint_policy_name: str 375 :keyword callable cls: A custom type or function that will be passed the direct response 376 :return: An iterator like instance of either ServiceEndpointPolicyDefinitionListResult or the result of cls(response) 377 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_03_01.models.ServiceEndpointPolicyDefinitionListResult] 378 :raises: ~azure.core.exceptions.HttpResponseError 379 """ 380 cls = kwargs.pop('cls', None) # type: ClsType["_models.ServiceEndpointPolicyDefinitionListResult"] 381 error_map = { 382 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 383 } 384 error_map.update(kwargs.pop('error_map', {})) 385 api_version = "2020-03-01" 386 accept = "application/json" 387 388 def prepare_request(next_link=None): 389 # Construct headers 390 header_parameters = {} # type: Dict[str, Any] 391 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 392 393 if not next_link: 394 # Construct URL 395 url = self.list_by_resource_group.metadata['url'] # type: ignore 396 path_format_arguments = { 397 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 398 'serviceEndpointPolicyName': self._serialize.url("service_endpoint_policy_name", service_endpoint_policy_name, 'str'), 399 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 400 } 401 url = self._client.format_url(url, **path_format_arguments) 402 # Construct parameters 403 query_parameters = {} # type: Dict[str, Any] 404 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 405 406 request = self._client.get(url, query_parameters, header_parameters) 407 else: 408 url = next_link 409 query_parameters = {} # type: Dict[str, Any] 410 request = self._client.get(url, query_parameters, header_parameters) 411 return request 412 413 async def extract_data(pipeline_response): 414 deserialized = self._deserialize('ServiceEndpointPolicyDefinitionListResult', pipeline_response) 415 list_of_elem = deserialized.value 416 if cls: 417 list_of_elem = cls(list_of_elem) 418 return deserialized.next_link or None, AsyncList(list_of_elem) 419 420 async def get_next(next_link=None): 421 request = prepare_request(next_link) 422 423 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 424 response = pipeline_response.http_response 425 426 if response.status_code not in [200]: 427 map_error(status_code=response.status_code, response=response, error_map=error_map) 428 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 429 430 return pipeline_response 431 432 return AsyncItemPaged( 433 get_next, extract_data 434 ) 435 list_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/serviceEndpointPolicies/{serviceEndpointPolicyName}/serviceEndpointPolicyDefinitions'} # type: ignore 436