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 PrivateEndpointsOperations: 25 """PrivateEndpointsOperations 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_07_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 private_endpoint_name: str, 50 **kwargs 51 ) -> None: 52 cls = kwargs.pop('cls', None) # type: ClsType[None] 53 error_map = { 54 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 55 } 56 error_map.update(kwargs.pop('error_map', {})) 57 api_version = "2020-07-01" 58 accept = "application/json" 59 60 # Construct URL 61 url = self._delete_initial.metadata['url'] # type: ignore 62 path_format_arguments = { 63 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 64 'privateEndpointName': self._serialize.url("private_endpoint_name", private_endpoint_name, 'str'), 65 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 66 } 67 url = self._client.format_url(url, **path_format_arguments) 68 69 # Construct parameters 70 query_parameters = {} # type: Dict[str, Any] 71 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 72 73 # Construct headers 74 header_parameters = {} # type: Dict[str, Any] 75 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 76 77 request = self._client.delete(url, query_parameters, header_parameters) 78 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 79 response = pipeline_response.http_response 80 81 if response.status_code not in [200, 202, 204]: 82 map_error(status_code=response.status_code, response=response, error_map=error_map) 83 error = self._deserialize.failsafe_deserialize(_models.Error, response) 84 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 85 86 if cls: 87 return cls(pipeline_response, None, {}) 88 89 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateEndpoints/{privateEndpointName}'} # type: ignore 90 91 async def begin_delete( 92 self, 93 resource_group_name: str, 94 private_endpoint_name: str, 95 **kwargs 96 ) -> AsyncLROPoller[None]: 97 """Deletes the specified private endpoint. 98 99 :param resource_group_name: The name of the resource group. 100 :type resource_group_name: str 101 :param private_endpoint_name: The name of the private endpoint. 102 :type private_endpoint_name: str 103 :keyword callable cls: A custom type or function that will be passed the direct response 104 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 105 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 106 False for no polling, or your own initialized polling object for a personal polling strategy. 107 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 108 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 109 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 110 :rtype: ~azure.core.polling.AsyncLROPoller[None] 111 :raises ~azure.core.exceptions.HttpResponseError: 112 """ 113 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 114 cls = kwargs.pop('cls', None) # type: ClsType[None] 115 lro_delay = kwargs.pop( 116 'polling_interval', 117 self._config.polling_interval 118 ) 119 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 120 if cont_token is None: 121 raw_result = await self._delete_initial( 122 resource_group_name=resource_group_name, 123 private_endpoint_name=private_endpoint_name, 124 cls=lambda x,y,z: x, 125 **kwargs 126 ) 127 128 kwargs.pop('error_map', None) 129 kwargs.pop('content_type', None) 130 131 def get_long_running_output(pipeline_response): 132 if cls: 133 return cls(pipeline_response, None, {}) 134 135 path_format_arguments = { 136 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 137 'privateEndpointName': self._serialize.url("private_endpoint_name", private_endpoint_name, 'str'), 138 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 139 } 140 141 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 142 elif polling is False: polling_method = AsyncNoPolling() 143 else: polling_method = polling 144 if cont_token: 145 return AsyncLROPoller.from_continuation_token( 146 polling_method=polling_method, 147 continuation_token=cont_token, 148 client=self._client, 149 deserialization_callback=get_long_running_output 150 ) 151 else: 152 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 153 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateEndpoints/{privateEndpointName}'} # type: ignore 154 155 async def get( 156 self, 157 resource_group_name: str, 158 private_endpoint_name: str, 159 expand: Optional[str] = None, 160 **kwargs 161 ) -> "_models.PrivateEndpoint": 162 """Gets the specified private endpoint by resource group. 163 164 :param resource_group_name: The name of the resource group. 165 :type resource_group_name: str 166 :param private_endpoint_name: The name of the private endpoint. 167 :type private_endpoint_name: str 168 :param expand: Expands referenced resources. 169 :type expand: str 170 :keyword callable cls: A custom type or function that will be passed the direct response 171 :return: PrivateEndpoint, or the result of cls(response) 172 :rtype: ~azure.mgmt.network.v2020_07_01.models.PrivateEndpoint 173 :raises: ~azure.core.exceptions.HttpResponseError 174 """ 175 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpoint"] 176 error_map = { 177 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 178 } 179 error_map.update(kwargs.pop('error_map', {})) 180 api_version = "2020-07-01" 181 accept = "application/json" 182 183 # Construct URL 184 url = self.get.metadata['url'] # type: ignore 185 path_format_arguments = { 186 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 187 'privateEndpointName': self._serialize.url("private_endpoint_name", private_endpoint_name, 'str'), 188 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 189 } 190 url = self._client.format_url(url, **path_format_arguments) 191 192 # Construct parameters 193 query_parameters = {} # type: Dict[str, Any] 194 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 195 if expand is not None: 196 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 197 198 # Construct headers 199 header_parameters = {} # type: Dict[str, Any] 200 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 201 202 request = self._client.get(url, query_parameters, header_parameters) 203 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 204 response = pipeline_response.http_response 205 206 if response.status_code not in [200]: 207 map_error(status_code=response.status_code, response=response, error_map=error_map) 208 error = self._deserialize.failsafe_deserialize(_models.Error, response) 209 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 210 211 deserialized = self._deserialize('PrivateEndpoint', pipeline_response) 212 213 if cls: 214 return cls(pipeline_response, deserialized, {}) 215 216 return deserialized 217 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateEndpoints/{privateEndpointName}'} # type: ignore 218 219 async def _create_or_update_initial( 220 self, 221 resource_group_name: str, 222 private_endpoint_name: str, 223 parameters: "_models.PrivateEndpoint", 224 **kwargs 225 ) -> "_models.PrivateEndpoint": 226 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpoint"] 227 error_map = { 228 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 229 } 230 error_map.update(kwargs.pop('error_map', {})) 231 api_version = "2020-07-01" 232 content_type = kwargs.pop("content_type", "application/json") 233 accept = "application/json" 234 235 # Construct URL 236 url = self._create_or_update_initial.metadata['url'] # type: ignore 237 path_format_arguments = { 238 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 239 'privateEndpointName': self._serialize.url("private_endpoint_name", private_endpoint_name, 'str'), 240 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 241 } 242 url = self._client.format_url(url, **path_format_arguments) 243 244 # Construct parameters 245 query_parameters = {} # type: Dict[str, Any] 246 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 247 248 # Construct headers 249 header_parameters = {} # type: Dict[str, Any] 250 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 251 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 252 253 body_content_kwargs = {} # type: Dict[str, Any] 254 body_content = self._serialize.body(parameters, 'PrivateEndpoint') 255 body_content_kwargs['content'] = body_content 256 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 257 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 258 response = pipeline_response.http_response 259 260 if response.status_code not in [200, 201]: 261 map_error(status_code=response.status_code, response=response, error_map=error_map) 262 error = self._deserialize.failsafe_deserialize(_models.Error, response) 263 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 264 265 if response.status_code == 200: 266 deserialized = self._deserialize('PrivateEndpoint', pipeline_response) 267 268 if response.status_code == 201: 269 deserialized = self._deserialize('PrivateEndpoint', pipeline_response) 270 271 if cls: 272 return cls(pipeline_response, deserialized, {}) 273 274 return deserialized 275 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateEndpoints/{privateEndpointName}'} # type: ignore 276 277 async def begin_create_or_update( 278 self, 279 resource_group_name: str, 280 private_endpoint_name: str, 281 parameters: "_models.PrivateEndpoint", 282 **kwargs 283 ) -> AsyncLROPoller["_models.PrivateEndpoint"]: 284 """Creates or updates an private endpoint in the specified resource group. 285 286 :param resource_group_name: The name of the resource group. 287 :type resource_group_name: str 288 :param private_endpoint_name: The name of the private endpoint. 289 :type private_endpoint_name: str 290 :param parameters: Parameters supplied to the create or update private endpoint operation. 291 :type parameters: ~azure.mgmt.network.v2020_07_01.models.PrivateEndpoint 292 :keyword callable cls: A custom type or function that will be passed the direct response 293 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 294 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 295 False for no polling, or your own initialized polling object for a personal polling strategy. 296 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 297 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 298 :return: An instance of AsyncLROPoller that returns either PrivateEndpoint or the result of cls(response) 299 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_07_01.models.PrivateEndpoint] 300 :raises ~azure.core.exceptions.HttpResponseError: 301 """ 302 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 303 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpoint"] 304 lro_delay = kwargs.pop( 305 'polling_interval', 306 self._config.polling_interval 307 ) 308 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 309 if cont_token is None: 310 raw_result = await self._create_or_update_initial( 311 resource_group_name=resource_group_name, 312 private_endpoint_name=private_endpoint_name, 313 parameters=parameters, 314 cls=lambda x,y,z: x, 315 **kwargs 316 ) 317 318 kwargs.pop('error_map', None) 319 kwargs.pop('content_type', None) 320 321 def get_long_running_output(pipeline_response): 322 deserialized = self._deserialize('PrivateEndpoint', pipeline_response) 323 324 if cls: 325 return cls(pipeline_response, deserialized, {}) 326 return deserialized 327 328 path_format_arguments = { 329 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 330 'privateEndpointName': self._serialize.url("private_endpoint_name", private_endpoint_name, 'str'), 331 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 332 } 333 334 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 335 elif polling is False: polling_method = AsyncNoPolling() 336 else: polling_method = polling 337 if cont_token: 338 return AsyncLROPoller.from_continuation_token( 339 polling_method=polling_method, 340 continuation_token=cont_token, 341 client=self._client, 342 deserialization_callback=get_long_running_output 343 ) 344 else: 345 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 346 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateEndpoints/{privateEndpointName}'} # type: ignore 347 348 def list( 349 self, 350 resource_group_name: str, 351 **kwargs 352 ) -> AsyncIterable["_models.PrivateEndpointListResult"]: 353 """Gets all private endpoints in a resource group. 354 355 :param resource_group_name: The name of the resource group. 356 :type resource_group_name: str 357 :keyword callable cls: A custom type or function that will be passed the direct response 358 :return: An iterator like instance of either PrivateEndpointListResult or the result of cls(response) 359 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_07_01.models.PrivateEndpointListResult] 360 :raises: ~azure.core.exceptions.HttpResponseError 361 """ 362 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpointListResult"] 363 error_map = { 364 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 365 } 366 error_map.update(kwargs.pop('error_map', {})) 367 api_version = "2020-07-01" 368 accept = "application/json" 369 370 def prepare_request(next_link=None): 371 # Construct headers 372 header_parameters = {} # type: Dict[str, Any] 373 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 374 375 if not next_link: 376 # Construct URL 377 url = self.list.metadata['url'] # type: ignore 378 path_format_arguments = { 379 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 380 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 381 } 382 url = self._client.format_url(url, **path_format_arguments) 383 # Construct parameters 384 query_parameters = {} # type: Dict[str, Any] 385 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 386 387 request = self._client.get(url, query_parameters, header_parameters) 388 else: 389 url = next_link 390 query_parameters = {} # type: Dict[str, Any] 391 request = self._client.get(url, query_parameters, header_parameters) 392 return request 393 394 async def extract_data(pipeline_response): 395 deserialized = self._deserialize('PrivateEndpointListResult', pipeline_response) 396 list_of_elem = deserialized.value 397 if cls: 398 list_of_elem = cls(list_of_elem) 399 return deserialized.next_link or None, AsyncList(list_of_elem) 400 401 async def get_next(next_link=None): 402 request = prepare_request(next_link) 403 404 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 405 response = pipeline_response.http_response 406 407 if response.status_code not in [200]: 408 error = self._deserialize.failsafe_deserialize(_models.Error, response) 409 map_error(status_code=response.status_code, response=response, error_map=error_map) 410 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 411 412 return pipeline_response 413 414 return AsyncItemPaged( 415 get_next, extract_data 416 ) 417 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateEndpoints'} # type: ignore 418 419 def list_by_subscription( 420 self, 421 **kwargs 422 ) -> AsyncIterable["_models.PrivateEndpointListResult"]: 423 """Gets all private endpoints in a subscription. 424 425 :keyword callable cls: A custom type or function that will be passed the direct response 426 :return: An iterator like instance of either PrivateEndpointListResult or the result of cls(response) 427 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_07_01.models.PrivateEndpointListResult] 428 :raises: ~azure.core.exceptions.HttpResponseError 429 """ 430 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpointListResult"] 431 error_map = { 432 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 433 } 434 error_map.update(kwargs.pop('error_map', {})) 435 api_version = "2020-07-01" 436 accept = "application/json" 437 438 def prepare_request(next_link=None): 439 # Construct headers 440 header_parameters = {} # type: Dict[str, Any] 441 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 442 443 if not next_link: 444 # Construct URL 445 url = self.list_by_subscription.metadata['url'] # type: ignore 446 path_format_arguments = { 447 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 448 } 449 url = self._client.format_url(url, **path_format_arguments) 450 # Construct parameters 451 query_parameters = {} # type: Dict[str, Any] 452 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 453 454 request = self._client.get(url, query_parameters, header_parameters) 455 else: 456 url = next_link 457 query_parameters = {} # type: Dict[str, Any] 458 request = self._client.get(url, query_parameters, header_parameters) 459 return request 460 461 async def extract_data(pipeline_response): 462 deserialized = self._deserialize('PrivateEndpointListResult', pipeline_response) 463 list_of_elem = deserialized.value 464 if cls: 465 list_of_elem = cls(list_of_elem) 466 return deserialized.next_link or None, AsyncList(list_of_elem) 467 468 async def get_next(next_link=None): 469 request = prepare_request(next_link) 470 471 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 472 response = pipeline_response.http_response 473 474 if response.status_code not in [200]: 475 error = self._deserialize.failsafe_deserialize(_models.Error, response) 476 map_error(status_code=response.status_code, response=response, error_map=error_map) 477 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 478 479 return pipeline_response 480 481 return AsyncItemPaged( 482 get_next, extract_data 483 ) 484 list_by_subscription.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/privateEndpoints'} # type: ignore 485