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.core.polling import LROPoller, NoPolling, PollingMethod 16from azure.mgmt.core.exceptions import ARMErrorFormat 17from azure.mgmt.core.polling.arm_polling import ARMPolling 18 19from .. import models as _models 20 21if TYPE_CHECKING: 22 # pylint: disable=unused-import,ungrouped-imports 23 from typing import Any, Callable, Dict, Generic, Iterable, Optional, TypeVar, Union 24 25 T = TypeVar('T') 26 ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]] 27 28class PrivateLinkServicesOperations(object): 29 """PrivateLinkServicesOperations operations. 30 31 You should not instantiate this class directly. Instead, you should create a Client instance that 32 instantiates it for you and attaches it as an attribute. 33 34 :ivar models: Alias to model classes used in this operation group. 35 :type models: ~azure.mgmt.network.v2019_11_01.models 36 :param client: Client for service requests. 37 :param config: Configuration of service client. 38 :param serializer: An object model serializer. 39 :param deserializer: An object model deserializer. 40 """ 41 42 models = _models 43 44 def __init__(self, client, config, serializer, deserializer): 45 self._client = client 46 self._serialize = serializer 47 self._deserialize = deserializer 48 self._config = config 49 50 def _delete_initial( 51 self, 52 resource_group_name, # type: str 53 service_name, # type: str 54 **kwargs # type: Any 55 ): 56 # type: (...) -> None 57 cls = kwargs.pop('cls', None) # type: ClsType[None] 58 error_map = { 59 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 60 } 61 error_map.update(kwargs.pop('error_map', {})) 62 api_version = "2019-11-01" 63 accept = "application/json" 64 65 # Construct URL 66 url = self._delete_initial.metadata['url'] # type: ignore 67 path_format_arguments = { 68 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 69 'serviceName': self._serialize.url("service_name", service_name, 'str'), 70 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 71 } 72 url = self._client.format_url(url, **path_format_arguments) 73 74 # Construct parameters 75 query_parameters = {} # type: Dict[str, Any] 76 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 77 78 # Construct headers 79 header_parameters = {} # type: Dict[str, Any] 80 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 81 82 request = self._client.delete(url, query_parameters, header_parameters) 83 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 84 response = pipeline_response.http_response 85 86 if response.status_code not in [200, 202, 204]: 87 map_error(status_code=response.status_code, response=response, error_map=error_map) 88 error = self._deserialize.failsafe_deserialize(_models.Error, response) 89 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 90 91 if cls: 92 return cls(pipeline_response, None, {}) 93 94 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}'} # type: ignore 95 96 def begin_delete( 97 self, 98 resource_group_name, # type: str 99 service_name, # type: str 100 **kwargs # type: Any 101 ): 102 # type: (...) -> LROPoller[None] 103 """Deletes the specified private link service. 104 105 :param resource_group_name: The name of the resource group. 106 :type resource_group_name: str 107 :param service_name: The name of the private link service. 108 :type service_name: str 109 :keyword callable cls: A custom type or function that will be passed the direct response 110 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 111 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 112 False for no polling, or your own initialized polling object for a personal polling strategy. 113 :paramtype polling: bool or ~azure.core.polling.PollingMethod 114 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 115 :return: An instance of LROPoller that returns either None or the result of cls(response) 116 :rtype: ~azure.core.polling.LROPoller[None] 117 :raises ~azure.core.exceptions.HttpResponseError: 118 """ 119 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 120 cls = kwargs.pop('cls', None) # type: ClsType[None] 121 lro_delay = kwargs.pop( 122 'polling_interval', 123 self._config.polling_interval 124 ) 125 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 126 if cont_token is None: 127 raw_result = self._delete_initial( 128 resource_group_name=resource_group_name, 129 service_name=service_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 'serviceName': self._serialize.url("service_name", service_name, 'str'), 144 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 145 } 146 147 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 148 elif polling is False: polling_method = NoPolling() 149 else: polling_method = polling 150 if cont_token: 151 return LROPoller.from_continuation_token( 152 polling_method=polling_method, 153 continuation_token=cont_token, 154 client=self._client, 155 deserialization_callback=get_long_running_output 156 ) 157 else: 158 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 159 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}'} # type: ignore 160 161 def get( 162 self, 163 resource_group_name, # type: str 164 service_name, # type: str 165 expand=None, # type: Optional[str] 166 **kwargs # type: Any 167 ): 168 # type: (...) -> "_models.PrivateLinkService" 169 """Gets the specified private link service by resource group. 170 171 :param resource_group_name: The name of the resource group. 172 :type resource_group_name: str 173 :param service_name: The name of the private link service. 174 :type service_name: str 175 :param expand: Expands referenced resources. 176 :type expand: str 177 :keyword callable cls: A custom type or function that will be passed the direct response 178 :return: PrivateLinkService, or the result of cls(response) 179 :rtype: ~azure.mgmt.network.v2019_11_01.models.PrivateLinkService 180 :raises: ~azure.core.exceptions.HttpResponseError 181 """ 182 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkService"] 183 error_map = { 184 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 185 } 186 error_map.update(kwargs.pop('error_map', {})) 187 api_version = "2019-11-01" 188 accept = "application/json" 189 190 # Construct URL 191 url = self.get.metadata['url'] # type: ignore 192 path_format_arguments = { 193 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 194 'serviceName': self._serialize.url("service_name", service_name, 'str'), 195 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 196 } 197 url = self._client.format_url(url, **path_format_arguments) 198 199 # Construct parameters 200 query_parameters = {} # type: Dict[str, Any] 201 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 202 if expand is not None: 203 query_parameters['$expand'] = self._serialize.query("expand", expand, '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 = 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 error = self._deserialize.failsafe_deserialize(_models.Error, response) 216 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 217 218 deserialized = self._deserialize('PrivateLinkService', pipeline_response) 219 220 if cls: 221 return cls(pipeline_response, deserialized, {}) 222 223 return deserialized 224 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}'} # type: ignore 225 226 def _create_or_update_initial( 227 self, 228 resource_group_name, # type: str 229 service_name, # type: str 230 parameters, # type: "_models.PrivateLinkService" 231 **kwargs # type: Any 232 ): 233 # type: (...) -> "_models.PrivateLinkService" 234 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkService"] 235 error_map = { 236 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 237 } 238 error_map.update(kwargs.pop('error_map', {})) 239 api_version = "2019-11-01" 240 content_type = kwargs.pop("content_type", "application/json") 241 accept = "application/json" 242 243 # Construct URL 244 url = self._create_or_update_initial.metadata['url'] # type: ignore 245 path_format_arguments = { 246 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 247 'serviceName': self._serialize.url("service_name", service_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(parameters, 'PrivateLinkService') 263 body_content_kwargs['content'] = body_content 264 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 265 pipeline_response = 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 error = self._deserialize.failsafe_deserialize(_models.Error, response) 271 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 272 273 if response.status_code == 200: 274 deserialized = self._deserialize('PrivateLinkService', pipeline_response) 275 276 if response.status_code == 201: 277 deserialized = self._deserialize('PrivateLinkService', pipeline_response) 278 279 if cls: 280 return cls(pipeline_response, deserialized, {}) 281 282 return deserialized 283 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}'} # type: ignore 284 285 def begin_create_or_update( 286 self, 287 resource_group_name, # type: str 288 service_name, # type: str 289 parameters, # type: "_models.PrivateLinkService" 290 **kwargs # type: Any 291 ): 292 # type: (...) -> LROPoller["_models.PrivateLinkService"] 293 """Creates or updates an private link service in the specified resource group. 294 295 :param resource_group_name: The name of the resource group. 296 :type resource_group_name: str 297 :param service_name: The name of the private link service. 298 :type service_name: str 299 :param parameters: Parameters supplied to the create or update private link service operation. 300 :type parameters: ~azure.mgmt.network.v2019_11_01.models.PrivateLinkService 301 :keyword callable cls: A custom type or function that will be passed the direct response 302 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 303 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 304 False for no polling, or your own initialized polling object for a personal polling strategy. 305 :paramtype polling: bool or ~azure.core.polling.PollingMethod 306 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 307 :return: An instance of LROPoller that returns either PrivateLinkService or the result of cls(response) 308 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.network.v2019_11_01.models.PrivateLinkService] 309 :raises ~azure.core.exceptions.HttpResponseError: 310 """ 311 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 312 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkService"] 313 lro_delay = kwargs.pop( 314 'polling_interval', 315 self._config.polling_interval 316 ) 317 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 318 if cont_token is None: 319 raw_result = self._create_or_update_initial( 320 resource_group_name=resource_group_name, 321 service_name=service_name, 322 parameters=parameters, 323 cls=lambda x,y,z: x, 324 **kwargs 325 ) 326 327 kwargs.pop('error_map', None) 328 kwargs.pop('content_type', None) 329 330 def get_long_running_output(pipeline_response): 331 deserialized = self._deserialize('PrivateLinkService', pipeline_response) 332 333 if cls: 334 return cls(pipeline_response, deserialized, {}) 335 return deserialized 336 337 path_format_arguments = { 338 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 339 'serviceName': self._serialize.url("service_name", service_name, 'str'), 340 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 341 } 342 343 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 344 elif polling is False: polling_method = NoPolling() 345 else: polling_method = polling 346 if cont_token: 347 return LROPoller.from_continuation_token( 348 polling_method=polling_method, 349 continuation_token=cont_token, 350 client=self._client, 351 deserialization_callback=get_long_running_output 352 ) 353 else: 354 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 355 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}'} # type: ignore 356 357 def list( 358 self, 359 resource_group_name, # type: str 360 **kwargs # type: Any 361 ): 362 # type: (...) -> Iterable["_models.PrivateLinkServiceListResult"] 363 """Gets all private link services in a resource group. 364 365 :param resource_group_name: The name of the resource group. 366 :type resource_group_name: str 367 :keyword callable cls: A custom type or function that will be passed the direct response 368 :return: An iterator like instance of either PrivateLinkServiceListResult or the result of cls(response) 369 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_11_01.models.PrivateLinkServiceListResult] 370 :raises: ~azure.core.exceptions.HttpResponseError 371 """ 372 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkServiceListResult"] 373 error_map = { 374 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 375 } 376 error_map.update(kwargs.pop('error_map', {})) 377 api_version = "2019-11-01" 378 accept = "application/json" 379 380 def prepare_request(next_link=None): 381 # Construct headers 382 header_parameters = {} # type: Dict[str, Any] 383 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 384 385 if not next_link: 386 # Construct URL 387 url = self.list.metadata['url'] # type: ignore 388 path_format_arguments = { 389 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 390 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 391 } 392 url = self._client.format_url(url, **path_format_arguments) 393 # Construct parameters 394 query_parameters = {} # type: Dict[str, Any] 395 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 396 397 request = self._client.get(url, query_parameters, header_parameters) 398 else: 399 url = next_link 400 query_parameters = {} # type: Dict[str, Any] 401 request = self._client.get(url, query_parameters, header_parameters) 402 return request 403 404 def extract_data(pipeline_response): 405 deserialized = self._deserialize('PrivateLinkServiceListResult', pipeline_response) 406 list_of_elem = deserialized.value 407 if cls: 408 list_of_elem = cls(list_of_elem) 409 return deserialized.next_link or None, iter(list_of_elem) 410 411 def get_next(next_link=None): 412 request = prepare_request(next_link) 413 414 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 415 response = pipeline_response.http_response 416 417 if response.status_code not in [200]: 418 error = self._deserialize.failsafe_deserialize(_models.Error, response) 419 map_error(status_code=response.status_code, response=response, error_map=error_map) 420 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 421 422 return pipeline_response 423 424 return ItemPaged( 425 get_next, extract_data 426 ) 427 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices'} # type: ignore 428 429 def list_by_subscription( 430 self, 431 **kwargs # type: Any 432 ): 433 # type: (...) -> Iterable["_models.PrivateLinkServiceListResult"] 434 """Gets all private link service in a subscription. 435 436 :keyword callable cls: A custom type or function that will be passed the direct response 437 :return: An iterator like instance of either PrivateLinkServiceListResult or the result of cls(response) 438 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_11_01.models.PrivateLinkServiceListResult] 439 :raises: ~azure.core.exceptions.HttpResponseError 440 """ 441 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkServiceListResult"] 442 error_map = { 443 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 444 } 445 error_map.update(kwargs.pop('error_map', {})) 446 api_version = "2019-11-01" 447 accept = "application/json" 448 449 def prepare_request(next_link=None): 450 # Construct headers 451 header_parameters = {} # type: Dict[str, Any] 452 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 453 454 if not next_link: 455 # Construct URL 456 url = self.list_by_subscription.metadata['url'] # type: ignore 457 path_format_arguments = { 458 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 459 } 460 url = self._client.format_url(url, **path_format_arguments) 461 # Construct parameters 462 query_parameters = {} # type: Dict[str, Any] 463 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 464 465 request = self._client.get(url, query_parameters, header_parameters) 466 else: 467 url = next_link 468 query_parameters = {} # type: Dict[str, Any] 469 request = self._client.get(url, query_parameters, header_parameters) 470 return request 471 472 def extract_data(pipeline_response): 473 deserialized = self._deserialize('PrivateLinkServiceListResult', pipeline_response) 474 list_of_elem = deserialized.value 475 if cls: 476 list_of_elem = cls(list_of_elem) 477 return deserialized.next_link or None, iter(list_of_elem) 478 479 def get_next(next_link=None): 480 request = prepare_request(next_link) 481 482 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 483 response = pipeline_response.http_response 484 485 if response.status_code not in [200]: 486 error = self._deserialize.failsafe_deserialize(_models.Error, response) 487 map_error(status_code=response.status_code, response=response, error_map=error_map) 488 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 489 490 return pipeline_response 491 492 return ItemPaged( 493 get_next, extract_data 494 ) 495 list_by_subscription.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/privateLinkServices'} # type: ignore 496 497 def get_private_endpoint_connection( 498 self, 499 resource_group_name, # type: str 500 service_name, # type: str 501 pe_connection_name, # type: str 502 expand=None, # type: Optional[str] 503 **kwargs # type: Any 504 ): 505 # type: (...) -> "_models.PrivateEndpointConnection" 506 """Get the specific private end point connection by specific private link service in the resource 507 group. 508 509 :param resource_group_name: The name of the resource group. 510 :type resource_group_name: str 511 :param service_name: The name of the private link service. 512 :type service_name: str 513 :param pe_connection_name: The name of the private end point connection. 514 :type pe_connection_name: str 515 :param expand: Expands referenced resources. 516 :type expand: str 517 :keyword callable cls: A custom type or function that will be passed the direct response 518 :return: PrivateEndpointConnection, or the result of cls(response) 519 :rtype: ~azure.mgmt.network.v2019_11_01.models.PrivateEndpointConnection 520 :raises: ~azure.core.exceptions.HttpResponseError 521 """ 522 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpointConnection"] 523 error_map = { 524 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 525 } 526 error_map.update(kwargs.pop('error_map', {})) 527 api_version = "2019-11-01" 528 accept = "application/json" 529 530 # Construct URL 531 url = self.get_private_endpoint_connection.metadata['url'] # type: ignore 532 path_format_arguments = { 533 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 534 'serviceName': self._serialize.url("service_name", service_name, 'str'), 535 'peConnectionName': self._serialize.url("pe_connection_name", pe_connection_name, 'str'), 536 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 537 } 538 url = self._client.format_url(url, **path_format_arguments) 539 540 # Construct parameters 541 query_parameters = {} # type: Dict[str, Any] 542 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 543 if expand is not None: 544 query_parameters['$expand'] = self._serialize.query("expand", expand, 'str') 545 546 # Construct headers 547 header_parameters = {} # type: Dict[str, Any] 548 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 549 550 request = self._client.get(url, query_parameters, header_parameters) 551 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 552 response = pipeline_response.http_response 553 554 if response.status_code not in [200]: 555 map_error(status_code=response.status_code, response=response, error_map=error_map) 556 error = self._deserialize.failsafe_deserialize(_models.Error, response) 557 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 558 559 deserialized = self._deserialize('PrivateEndpointConnection', pipeline_response) 560 561 if cls: 562 return cls(pipeline_response, deserialized, {}) 563 564 return deserialized 565 get_private_endpoint_connection.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}/privateEndpointConnections/{peConnectionName}'} # type: ignore 566 567 def update_private_endpoint_connection( 568 self, 569 resource_group_name, # type: str 570 service_name, # type: str 571 pe_connection_name, # type: str 572 parameters, # type: "_models.PrivateEndpointConnection" 573 **kwargs # type: Any 574 ): 575 # type: (...) -> "_models.PrivateEndpointConnection" 576 """Approve or reject private end point connection for a private link service in a subscription. 577 578 :param resource_group_name: The name of the resource group. 579 :type resource_group_name: str 580 :param service_name: The name of the private link service. 581 :type service_name: str 582 :param pe_connection_name: The name of the private end point connection. 583 :type pe_connection_name: str 584 :param parameters: Parameters supplied to approve or reject the private end point connection. 585 :type parameters: ~azure.mgmt.network.v2019_11_01.models.PrivateEndpointConnection 586 :keyword callable cls: A custom type or function that will be passed the direct response 587 :return: PrivateEndpointConnection, or the result of cls(response) 588 :rtype: ~azure.mgmt.network.v2019_11_01.models.PrivateEndpointConnection 589 :raises: ~azure.core.exceptions.HttpResponseError 590 """ 591 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpointConnection"] 592 error_map = { 593 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 594 } 595 error_map.update(kwargs.pop('error_map', {})) 596 api_version = "2019-11-01" 597 content_type = kwargs.pop("content_type", "application/json") 598 accept = "application/json" 599 600 # Construct URL 601 url = self.update_private_endpoint_connection.metadata['url'] # type: ignore 602 path_format_arguments = { 603 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 604 'serviceName': self._serialize.url("service_name", service_name, 'str'), 605 'peConnectionName': self._serialize.url("pe_connection_name", pe_connection_name, 'str'), 606 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 607 } 608 url = self._client.format_url(url, **path_format_arguments) 609 610 # Construct parameters 611 query_parameters = {} # type: Dict[str, Any] 612 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 613 614 # Construct headers 615 header_parameters = {} # type: Dict[str, Any] 616 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 617 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 618 619 body_content_kwargs = {} # type: Dict[str, Any] 620 body_content = self._serialize.body(parameters, 'PrivateEndpointConnection') 621 body_content_kwargs['content'] = body_content 622 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 623 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 624 response = pipeline_response.http_response 625 626 if response.status_code not in [200]: 627 map_error(status_code=response.status_code, response=response, error_map=error_map) 628 error = self._deserialize.failsafe_deserialize(_models.Error, response) 629 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 630 631 deserialized = self._deserialize('PrivateEndpointConnection', pipeline_response) 632 633 if cls: 634 return cls(pipeline_response, deserialized, {}) 635 636 return deserialized 637 update_private_endpoint_connection.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}/privateEndpointConnections/{peConnectionName}'} # type: ignore 638 639 def _delete_private_endpoint_connection_initial( 640 self, 641 resource_group_name, # type: str 642 service_name, # type: str 643 pe_connection_name, # type: str 644 **kwargs # type: Any 645 ): 646 # type: (...) -> None 647 cls = kwargs.pop('cls', None) # type: ClsType[None] 648 error_map = { 649 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 650 } 651 error_map.update(kwargs.pop('error_map', {})) 652 api_version = "2019-11-01" 653 accept = "application/json" 654 655 # Construct URL 656 url = self._delete_private_endpoint_connection_initial.metadata['url'] # type: ignore 657 path_format_arguments = { 658 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 659 'serviceName': self._serialize.url("service_name", service_name, 'str'), 660 'peConnectionName': self._serialize.url("pe_connection_name", pe_connection_name, 'str'), 661 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 662 } 663 url = self._client.format_url(url, **path_format_arguments) 664 665 # Construct parameters 666 query_parameters = {} # type: Dict[str, Any] 667 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 668 669 # Construct headers 670 header_parameters = {} # type: Dict[str, Any] 671 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 672 673 request = self._client.delete(url, query_parameters, header_parameters) 674 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 675 response = pipeline_response.http_response 676 677 if response.status_code not in [200, 202, 204]: 678 map_error(status_code=response.status_code, response=response, error_map=error_map) 679 error = self._deserialize.failsafe_deserialize(_models.Error, response) 680 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 681 682 if cls: 683 return cls(pipeline_response, None, {}) 684 685 _delete_private_endpoint_connection_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}/privateEndpointConnections/{peConnectionName}'} # type: ignore 686 687 def begin_delete_private_endpoint_connection( 688 self, 689 resource_group_name, # type: str 690 service_name, # type: str 691 pe_connection_name, # type: str 692 **kwargs # type: Any 693 ): 694 # type: (...) -> LROPoller[None] 695 """Delete private end point connection for a private link service in a subscription. 696 697 :param resource_group_name: The name of the resource group. 698 :type resource_group_name: str 699 :param service_name: The name of the private link service. 700 :type service_name: str 701 :param pe_connection_name: The name of the private end point connection. 702 :type pe_connection_name: str 703 :keyword callable cls: A custom type or function that will be passed the direct response 704 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 705 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 706 False for no polling, or your own initialized polling object for a personal polling strategy. 707 :paramtype polling: bool or ~azure.core.polling.PollingMethod 708 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 709 :return: An instance of LROPoller that returns either None or the result of cls(response) 710 :rtype: ~azure.core.polling.LROPoller[None] 711 :raises ~azure.core.exceptions.HttpResponseError: 712 """ 713 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 714 cls = kwargs.pop('cls', None) # type: ClsType[None] 715 lro_delay = kwargs.pop( 716 'polling_interval', 717 self._config.polling_interval 718 ) 719 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 720 if cont_token is None: 721 raw_result = self._delete_private_endpoint_connection_initial( 722 resource_group_name=resource_group_name, 723 service_name=service_name, 724 pe_connection_name=pe_connection_name, 725 cls=lambda x,y,z: x, 726 **kwargs 727 ) 728 729 kwargs.pop('error_map', None) 730 kwargs.pop('content_type', None) 731 732 def get_long_running_output(pipeline_response): 733 if cls: 734 return cls(pipeline_response, None, {}) 735 736 path_format_arguments = { 737 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 738 'serviceName': self._serialize.url("service_name", service_name, 'str'), 739 'peConnectionName': self._serialize.url("pe_connection_name", pe_connection_name, 'str'), 740 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 741 } 742 743 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 744 elif polling is False: polling_method = NoPolling() 745 else: polling_method = polling 746 if cont_token: 747 return LROPoller.from_continuation_token( 748 polling_method=polling_method, 749 continuation_token=cont_token, 750 client=self._client, 751 deserialization_callback=get_long_running_output 752 ) 753 else: 754 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 755 begin_delete_private_endpoint_connection.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}/privateEndpointConnections/{peConnectionName}'} # type: ignore 756 757 def list_private_endpoint_connections( 758 self, 759 resource_group_name, # type: str 760 service_name, # type: str 761 **kwargs # type: Any 762 ): 763 # type: (...) -> Iterable["_models.PrivateEndpointConnectionListResult"] 764 """Gets all private end point connections for a specific private link service. 765 766 :param resource_group_name: The name of the resource group. 767 :type resource_group_name: str 768 :param service_name: The name of the private link service. 769 :type service_name: str 770 :keyword callable cls: A custom type or function that will be passed the direct response 771 :return: An iterator like instance of either PrivateEndpointConnectionListResult or the result of cls(response) 772 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_11_01.models.PrivateEndpointConnectionListResult] 773 :raises: ~azure.core.exceptions.HttpResponseError 774 """ 775 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpointConnectionListResult"] 776 error_map = { 777 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 778 } 779 error_map.update(kwargs.pop('error_map', {})) 780 api_version = "2019-11-01" 781 accept = "application/json" 782 783 def prepare_request(next_link=None): 784 # Construct headers 785 header_parameters = {} # type: Dict[str, Any] 786 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 787 788 if not next_link: 789 # Construct URL 790 url = self.list_private_endpoint_connections.metadata['url'] # type: ignore 791 path_format_arguments = { 792 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 793 'serviceName': self._serialize.url("service_name", service_name, 'str'), 794 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 795 } 796 url = self._client.format_url(url, **path_format_arguments) 797 # Construct parameters 798 query_parameters = {} # type: Dict[str, Any] 799 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 800 801 request = self._client.get(url, query_parameters, header_parameters) 802 else: 803 url = next_link 804 query_parameters = {} # type: Dict[str, Any] 805 request = self._client.get(url, query_parameters, header_parameters) 806 return request 807 808 def extract_data(pipeline_response): 809 deserialized = self._deserialize('PrivateEndpointConnectionListResult', pipeline_response) 810 list_of_elem = deserialized.value 811 if cls: 812 list_of_elem = cls(list_of_elem) 813 return deserialized.next_link or None, iter(list_of_elem) 814 815 def get_next(next_link=None): 816 request = prepare_request(next_link) 817 818 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 819 response = pipeline_response.http_response 820 821 if response.status_code not in [200]: 822 error = self._deserialize.failsafe_deserialize(_models.Error, response) 823 map_error(status_code=response.status_code, response=response, error_map=error_map) 824 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 825 826 return pipeline_response 827 828 return ItemPaged( 829 get_next, extract_data 830 ) 831 list_private_endpoint_connections.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}/privateEndpointConnections'} # type: ignore 832 833 def _check_private_link_service_visibility_initial( 834 self, 835 location, # type: str 836 parameters, # type: "_models.CheckPrivateLinkServiceVisibilityRequest" 837 **kwargs # type: Any 838 ): 839 # type: (...) -> Optional["_models.PrivateLinkServiceVisibility"] 840 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.PrivateLinkServiceVisibility"]] 841 error_map = { 842 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 843 } 844 error_map.update(kwargs.pop('error_map', {})) 845 api_version = "2019-11-01" 846 content_type = kwargs.pop("content_type", "application/json") 847 accept = "application/json" 848 849 # Construct URL 850 url = self._check_private_link_service_visibility_initial.metadata['url'] # type: ignore 851 path_format_arguments = { 852 'location': self._serialize.url("location", location, 'str'), 853 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 854 } 855 url = self._client.format_url(url, **path_format_arguments) 856 857 # Construct parameters 858 query_parameters = {} # type: Dict[str, Any] 859 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 860 861 # Construct headers 862 header_parameters = {} # type: Dict[str, Any] 863 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 864 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 865 866 body_content_kwargs = {} # type: Dict[str, Any] 867 body_content = self._serialize.body(parameters, 'CheckPrivateLinkServiceVisibilityRequest') 868 body_content_kwargs['content'] = body_content 869 request = self._client.post(url, query_parameters, header_parameters, **body_content_kwargs) 870 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 871 response = pipeline_response.http_response 872 873 if response.status_code not in [200, 202]: 874 map_error(status_code=response.status_code, response=response, error_map=error_map) 875 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 876 877 deserialized = None 878 if response.status_code == 200: 879 deserialized = self._deserialize('PrivateLinkServiceVisibility', pipeline_response) 880 881 if cls: 882 return cls(pipeline_response, deserialized, {}) 883 884 return deserialized 885 _check_private_link_service_visibility_initial.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/locations/{location}/checkPrivateLinkServiceVisibility'} # type: ignore 886 887 def begin_check_private_link_service_visibility( 888 self, 889 location, # type: str 890 parameters, # type: "_models.CheckPrivateLinkServiceVisibilityRequest" 891 **kwargs # type: Any 892 ): 893 # type: (...) -> LROPoller["_models.PrivateLinkServiceVisibility"] 894 """Checks whether the subscription is visible to private link service. 895 896 :param location: The location of the domain name. 897 :type location: str 898 :param parameters: The request body of CheckPrivateLinkService API call. 899 :type parameters: ~azure.mgmt.network.v2019_11_01.models.CheckPrivateLinkServiceVisibilityRequest 900 :keyword callable cls: A custom type or function that will be passed the direct response 901 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 902 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 903 False for no polling, or your own initialized polling object for a personal polling strategy. 904 :paramtype polling: bool or ~azure.core.polling.PollingMethod 905 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 906 :return: An instance of LROPoller that returns either PrivateLinkServiceVisibility or the result of cls(response) 907 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.network.v2019_11_01.models.PrivateLinkServiceVisibility] 908 :raises ~azure.core.exceptions.HttpResponseError: 909 """ 910 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 911 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkServiceVisibility"] 912 lro_delay = kwargs.pop( 913 'polling_interval', 914 self._config.polling_interval 915 ) 916 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 917 if cont_token is None: 918 raw_result = self._check_private_link_service_visibility_initial( 919 location=location, 920 parameters=parameters, 921 cls=lambda x,y,z: x, 922 **kwargs 923 ) 924 925 kwargs.pop('error_map', None) 926 kwargs.pop('content_type', None) 927 928 def get_long_running_output(pipeline_response): 929 deserialized = self._deserialize('PrivateLinkServiceVisibility', pipeline_response) 930 931 if cls: 932 return cls(pipeline_response, deserialized, {}) 933 return deserialized 934 935 path_format_arguments = { 936 'location': self._serialize.url("location", location, 'str'), 937 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 938 } 939 940 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 941 elif polling is False: polling_method = NoPolling() 942 else: polling_method = polling 943 if cont_token: 944 return LROPoller.from_continuation_token( 945 polling_method=polling_method, 946 continuation_token=cont_token, 947 client=self._client, 948 deserialization_callback=get_long_running_output 949 ) 950 else: 951 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 952 begin_check_private_link_service_visibility.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/locations/{location}/checkPrivateLinkServiceVisibility'} # type: ignore 953 954 def _check_private_link_service_visibility_by_resource_group_initial( 955 self, 956 location, # type: str 957 resource_group_name, # type: str 958 parameters, # type: "_models.CheckPrivateLinkServiceVisibilityRequest" 959 **kwargs # type: Any 960 ): 961 # type: (...) -> Optional["_models.PrivateLinkServiceVisibility"] 962 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.PrivateLinkServiceVisibility"]] 963 error_map = { 964 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 965 } 966 error_map.update(kwargs.pop('error_map', {})) 967 api_version = "2019-11-01" 968 content_type = kwargs.pop("content_type", "application/json") 969 accept = "application/json" 970 971 # Construct URL 972 url = self._check_private_link_service_visibility_by_resource_group_initial.metadata['url'] # type: ignore 973 path_format_arguments = { 974 'location': self._serialize.url("location", location, 'str'), 975 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 976 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 977 } 978 url = self._client.format_url(url, **path_format_arguments) 979 980 # Construct parameters 981 query_parameters = {} # type: Dict[str, Any] 982 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 983 984 # Construct headers 985 header_parameters = {} # type: Dict[str, Any] 986 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 987 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 988 989 body_content_kwargs = {} # type: Dict[str, Any] 990 body_content = self._serialize.body(parameters, 'CheckPrivateLinkServiceVisibilityRequest') 991 body_content_kwargs['content'] = body_content 992 request = self._client.post(url, query_parameters, header_parameters, **body_content_kwargs) 993 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 994 response = pipeline_response.http_response 995 996 if response.status_code not in [200, 202]: 997 map_error(status_code=response.status_code, response=response, error_map=error_map) 998 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 999 1000 deserialized = None 1001 if response.status_code == 200: 1002 deserialized = self._deserialize('PrivateLinkServiceVisibility', pipeline_response) 1003 1004 if cls: 1005 return cls(pipeline_response, deserialized, {}) 1006 1007 return deserialized 1008 _check_private_link_service_visibility_by_resource_group_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/locations/{location}/checkPrivateLinkServiceVisibility'} # type: ignore 1009 1010 def begin_check_private_link_service_visibility_by_resource_group( 1011 self, 1012 location, # type: str 1013 resource_group_name, # type: str 1014 parameters, # type: "_models.CheckPrivateLinkServiceVisibilityRequest" 1015 **kwargs # type: Any 1016 ): 1017 # type: (...) -> LROPoller["_models.PrivateLinkServiceVisibility"] 1018 """Checks whether the subscription is visible to private link service in the specified resource 1019 group. 1020 1021 :param location: The location of the domain name. 1022 :type location: str 1023 :param resource_group_name: The name of the resource group. 1024 :type resource_group_name: str 1025 :param parameters: The request body of CheckPrivateLinkService API call. 1026 :type parameters: ~azure.mgmt.network.v2019_11_01.models.CheckPrivateLinkServiceVisibilityRequest 1027 :keyword callable cls: A custom type or function that will be passed the direct response 1028 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 1029 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 1030 False for no polling, or your own initialized polling object for a personal polling strategy. 1031 :paramtype polling: bool or ~azure.core.polling.PollingMethod 1032 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 1033 :return: An instance of LROPoller that returns either PrivateLinkServiceVisibility or the result of cls(response) 1034 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.network.v2019_11_01.models.PrivateLinkServiceVisibility] 1035 :raises ~azure.core.exceptions.HttpResponseError: 1036 """ 1037 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 1038 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkServiceVisibility"] 1039 lro_delay = kwargs.pop( 1040 'polling_interval', 1041 self._config.polling_interval 1042 ) 1043 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 1044 if cont_token is None: 1045 raw_result = self._check_private_link_service_visibility_by_resource_group_initial( 1046 location=location, 1047 resource_group_name=resource_group_name, 1048 parameters=parameters, 1049 cls=lambda x,y,z: x, 1050 **kwargs 1051 ) 1052 1053 kwargs.pop('error_map', None) 1054 kwargs.pop('content_type', None) 1055 1056 def get_long_running_output(pipeline_response): 1057 deserialized = self._deserialize('PrivateLinkServiceVisibility', pipeline_response) 1058 1059 if cls: 1060 return cls(pipeline_response, deserialized, {}) 1061 return deserialized 1062 1063 path_format_arguments = { 1064 'location': self._serialize.url("location", location, 'str'), 1065 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 1066 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 1067 } 1068 1069 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 1070 elif polling is False: polling_method = NoPolling() 1071 else: polling_method = polling 1072 if cont_token: 1073 return LROPoller.from_continuation_token( 1074 polling_method=polling_method, 1075 continuation_token=cont_token, 1076 client=self._client, 1077 deserialization_callback=get_long_running_output 1078 ) 1079 else: 1080 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 1081 begin_check_private_link_service_visibility_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/locations/{location}/checkPrivateLinkServiceVisibility'} # type: ignore 1082 1083 def list_auto_approved_private_link_services( 1084 self, 1085 location, # type: str 1086 **kwargs # type: Any 1087 ): 1088 # type: (...) -> Iterable["_models.AutoApprovedPrivateLinkServicesResult"] 1089 """Returns all of the private link service ids that can be linked to a Private Endpoint with auto 1090 approved in this subscription in this region. 1091 1092 :param location: The location of the domain name. 1093 :type location: str 1094 :keyword callable cls: A custom type or function that will be passed the direct response 1095 :return: An iterator like instance of either AutoApprovedPrivateLinkServicesResult or the result of cls(response) 1096 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_11_01.models.AutoApprovedPrivateLinkServicesResult] 1097 :raises: ~azure.core.exceptions.HttpResponseError 1098 """ 1099 cls = kwargs.pop('cls', None) # type: ClsType["_models.AutoApprovedPrivateLinkServicesResult"] 1100 error_map = { 1101 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1102 } 1103 error_map.update(kwargs.pop('error_map', {})) 1104 api_version = "2019-11-01" 1105 accept = "application/json" 1106 1107 def prepare_request(next_link=None): 1108 # Construct headers 1109 header_parameters = {} # type: Dict[str, Any] 1110 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1111 1112 if not next_link: 1113 # Construct URL 1114 url = self.list_auto_approved_private_link_services.metadata['url'] # type: ignore 1115 path_format_arguments = { 1116 'location': self._serialize.url("location", location, 'str'), 1117 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 1118 } 1119 url = self._client.format_url(url, **path_format_arguments) 1120 # Construct parameters 1121 query_parameters = {} # type: Dict[str, Any] 1122 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1123 1124 request = self._client.get(url, query_parameters, header_parameters) 1125 else: 1126 url = next_link 1127 query_parameters = {} # type: Dict[str, Any] 1128 request = self._client.get(url, query_parameters, header_parameters) 1129 return request 1130 1131 def extract_data(pipeline_response): 1132 deserialized = self._deserialize('AutoApprovedPrivateLinkServicesResult', pipeline_response) 1133 list_of_elem = deserialized.value 1134 if cls: 1135 list_of_elem = cls(list_of_elem) 1136 return deserialized.next_link or None, iter(list_of_elem) 1137 1138 def get_next(next_link=None): 1139 request = prepare_request(next_link) 1140 1141 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1142 response = pipeline_response.http_response 1143 1144 if response.status_code not in [200]: 1145 map_error(status_code=response.status_code, response=response, error_map=error_map) 1146 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1147 1148 return pipeline_response 1149 1150 return ItemPaged( 1151 get_next, extract_data 1152 ) 1153 list_auto_approved_private_link_services.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/locations/{location}/autoApprovedPrivateLinkServices'} # type: ignore 1154 1155 def list_auto_approved_private_link_services_by_resource_group( 1156 self, 1157 location, # type: str 1158 resource_group_name, # type: str 1159 **kwargs # type: Any 1160 ): 1161 # type: (...) -> Iterable["_models.AutoApprovedPrivateLinkServicesResult"] 1162 """Returns all of the private link service ids that can be linked to a Private Endpoint with auto 1163 approved in this subscription in this region. 1164 1165 :param location: The location of the domain name. 1166 :type location: str 1167 :param resource_group_name: The name of the resource group. 1168 :type resource_group_name: str 1169 :keyword callable cls: A custom type or function that will be passed the direct response 1170 :return: An iterator like instance of either AutoApprovedPrivateLinkServicesResult or the result of cls(response) 1171 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_11_01.models.AutoApprovedPrivateLinkServicesResult] 1172 :raises: ~azure.core.exceptions.HttpResponseError 1173 """ 1174 cls = kwargs.pop('cls', None) # type: ClsType["_models.AutoApprovedPrivateLinkServicesResult"] 1175 error_map = { 1176 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1177 } 1178 error_map.update(kwargs.pop('error_map', {})) 1179 api_version = "2019-11-01" 1180 accept = "application/json" 1181 1182 def prepare_request(next_link=None): 1183 # Construct headers 1184 header_parameters = {} # type: Dict[str, Any] 1185 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1186 1187 if not next_link: 1188 # Construct URL 1189 url = self.list_auto_approved_private_link_services_by_resource_group.metadata['url'] # type: ignore 1190 path_format_arguments = { 1191 'location': self._serialize.url("location", location, 'str'), 1192 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 1193 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 1194 } 1195 url = self._client.format_url(url, **path_format_arguments) 1196 # Construct parameters 1197 query_parameters = {} # type: Dict[str, Any] 1198 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1199 1200 request = self._client.get(url, query_parameters, header_parameters) 1201 else: 1202 url = next_link 1203 query_parameters = {} # type: Dict[str, Any] 1204 request = self._client.get(url, query_parameters, header_parameters) 1205 return request 1206 1207 def extract_data(pipeline_response): 1208 deserialized = self._deserialize('AutoApprovedPrivateLinkServicesResult', pipeline_response) 1209 list_of_elem = deserialized.value 1210 if cls: 1211 list_of_elem = cls(list_of_elem) 1212 return deserialized.next_link or None, iter(list_of_elem) 1213 1214 def get_next(next_link=None): 1215 request = prepare_request(next_link) 1216 1217 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1218 response = pipeline_response.http_response 1219 1220 if response.status_code not in [200]: 1221 map_error(status_code=response.status_code, response=response, error_map=error_map) 1222 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1223 1224 return pipeline_response 1225 1226 return ItemPaged( 1227 get_next, extract_data 1228 ) 1229 list_auto_approved_private_link_services_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/locations/{location}/autoApprovedPrivateLinkServices'} # type: ignore 1230