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_07_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-07-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_07_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-07-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-07-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_07_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_07_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_07_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-07-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_07_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-07-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 update_private_endpoint_connection( 498 self, 499 resource_group_name, # type: str 500 service_name, # type: str 501 pe_connection_name, # type: str 502 parameters, # type: "_models.PrivateEndpointConnection" 503 **kwargs # type: Any 504 ): 505 # type: (...) -> "_models.PrivateEndpointConnection" 506 """Approve or reject private end point connection for a private link service in a subscription. 507 508 :param resource_group_name: The name of the resource group. 509 :type resource_group_name: str 510 :param service_name: The name of the private link service. 511 :type service_name: str 512 :param pe_connection_name: The name of the private end point connection. 513 :type pe_connection_name: str 514 :param parameters: Parameters supplied to approve or reject the private end point connection. 515 :type parameters: ~azure.mgmt.network.v2019_07_01.models.PrivateEndpointConnection 516 :keyword callable cls: A custom type or function that will be passed the direct response 517 :return: PrivateEndpointConnection, or the result of cls(response) 518 :rtype: ~azure.mgmt.network.v2019_07_01.models.PrivateEndpointConnection 519 :raises: ~azure.core.exceptions.HttpResponseError 520 """ 521 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateEndpointConnection"] 522 error_map = { 523 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 524 } 525 error_map.update(kwargs.pop('error_map', {})) 526 api_version = "2019-07-01" 527 content_type = kwargs.pop("content_type", "application/json") 528 accept = "application/json" 529 530 # Construct URL 531 url = self.update_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 544 # Construct headers 545 header_parameters = {} # type: Dict[str, Any] 546 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 547 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 548 549 body_content_kwargs = {} # type: Dict[str, Any] 550 body_content = self._serialize.body(parameters, 'PrivateEndpointConnection') 551 body_content_kwargs['content'] = body_content 552 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 553 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 554 response = pipeline_response.http_response 555 556 if response.status_code not in [200]: 557 map_error(status_code=response.status_code, response=response, error_map=error_map) 558 error = self._deserialize.failsafe_deserialize(_models.Error, response) 559 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 560 561 deserialized = self._deserialize('PrivateEndpointConnection', pipeline_response) 562 563 if cls: 564 return cls(pipeline_response, deserialized, {}) 565 566 return deserialized 567 update_private_endpoint_connection.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}/privateEndpointConnections/{peConnectionName}'} # type: ignore 568 569 def _delete_private_endpoint_connection_initial( 570 self, 571 resource_group_name, # type: str 572 service_name, # type: str 573 pe_connection_name, # type: str 574 **kwargs # type: Any 575 ): 576 # type: (...) -> None 577 cls = kwargs.pop('cls', None) # type: ClsType[None] 578 error_map = { 579 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 580 } 581 error_map.update(kwargs.pop('error_map', {})) 582 api_version = "2019-07-01" 583 accept = "application/json" 584 585 # Construct URL 586 url = self._delete_private_endpoint_connection_initial.metadata['url'] # type: ignore 587 path_format_arguments = { 588 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 589 'serviceName': self._serialize.url("service_name", service_name, 'str'), 590 'peConnectionName': self._serialize.url("pe_connection_name", pe_connection_name, 'str'), 591 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 592 } 593 url = self._client.format_url(url, **path_format_arguments) 594 595 # Construct parameters 596 query_parameters = {} # type: Dict[str, Any] 597 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 598 599 # Construct headers 600 header_parameters = {} # type: Dict[str, Any] 601 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 602 603 request = self._client.delete(url, query_parameters, header_parameters) 604 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 605 response = pipeline_response.http_response 606 607 if response.status_code not in [200, 202, 204]: 608 map_error(status_code=response.status_code, response=response, error_map=error_map) 609 error = self._deserialize.failsafe_deserialize(_models.Error, response) 610 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 611 612 if cls: 613 return cls(pipeline_response, None, {}) 614 615 _delete_private_endpoint_connection_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}/privateEndpointConnections/{peConnectionName}'} # type: ignore 616 617 def begin_delete_private_endpoint_connection( 618 self, 619 resource_group_name, # type: str 620 service_name, # type: str 621 pe_connection_name, # type: str 622 **kwargs # type: Any 623 ): 624 # type: (...) -> LROPoller[None] 625 """Delete private end point connection for a private link service in a subscription. 626 627 :param resource_group_name: The name of the resource group. 628 :type resource_group_name: str 629 :param service_name: The name of the private link service. 630 :type service_name: str 631 :param pe_connection_name: The name of the private end point connection. 632 :type pe_connection_name: str 633 :keyword callable cls: A custom type or function that will be passed the direct response 634 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 635 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 636 False for no polling, or your own initialized polling object for a personal polling strategy. 637 :paramtype polling: bool or ~azure.core.polling.PollingMethod 638 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 639 :return: An instance of LROPoller that returns either None or the result of cls(response) 640 :rtype: ~azure.core.polling.LROPoller[None] 641 :raises ~azure.core.exceptions.HttpResponseError: 642 """ 643 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 644 cls = kwargs.pop('cls', None) # type: ClsType[None] 645 lro_delay = kwargs.pop( 646 'polling_interval', 647 self._config.polling_interval 648 ) 649 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 650 if cont_token is None: 651 raw_result = self._delete_private_endpoint_connection_initial( 652 resource_group_name=resource_group_name, 653 service_name=service_name, 654 pe_connection_name=pe_connection_name, 655 cls=lambda x,y,z: x, 656 **kwargs 657 ) 658 659 kwargs.pop('error_map', None) 660 kwargs.pop('content_type', None) 661 662 def get_long_running_output(pipeline_response): 663 if cls: 664 return cls(pipeline_response, None, {}) 665 666 path_format_arguments = { 667 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 668 'serviceName': self._serialize.url("service_name", service_name, 'str'), 669 'peConnectionName': self._serialize.url("pe_connection_name", pe_connection_name, 'str'), 670 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 671 } 672 673 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 674 elif polling is False: polling_method = NoPolling() 675 else: polling_method = polling 676 if cont_token: 677 return LROPoller.from_continuation_token( 678 polling_method=polling_method, 679 continuation_token=cont_token, 680 client=self._client, 681 deserialization_callback=get_long_running_output 682 ) 683 else: 684 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 685 begin_delete_private_endpoint_connection.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/privateLinkServices/{serviceName}/privateEndpointConnections/{peConnectionName}'} # type: ignore 686 687 def _check_private_link_service_visibility_initial( 688 self, 689 location, # type: str 690 parameters, # type: "_models.CheckPrivateLinkServiceVisibilityRequest" 691 **kwargs # type: Any 692 ): 693 # type: (...) -> Optional["_models.PrivateLinkServiceVisibility"] 694 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.PrivateLinkServiceVisibility"]] 695 error_map = { 696 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 697 } 698 error_map.update(kwargs.pop('error_map', {})) 699 api_version = "2019-07-01" 700 content_type = kwargs.pop("content_type", "application/json") 701 accept = "application/json" 702 703 # Construct URL 704 url = self._check_private_link_service_visibility_initial.metadata['url'] # type: ignore 705 path_format_arguments = { 706 'location': self._serialize.url("location", location, 'str'), 707 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 708 } 709 url = self._client.format_url(url, **path_format_arguments) 710 711 # Construct parameters 712 query_parameters = {} # type: Dict[str, Any] 713 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 714 715 # Construct headers 716 header_parameters = {} # type: Dict[str, Any] 717 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 718 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 719 720 body_content_kwargs = {} # type: Dict[str, Any] 721 body_content = self._serialize.body(parameters, 'CheckPrivateLinkServiceVisibilityRequest') 722 body_content_kwargs['content'] = body_content 723 request = self._client.post(url, query_parameters, header_parameters, **body_content_kwargs) 724 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 725 response = pipeline_response.http_response 726 727 if response.status_code not in [200, 202]: 728 map_error(status_code=response.status_code, response=response, error_map=error_map) 729 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 730 731 deserialized = None 732 if response.status_code == 200: 733 deserialized = self._deserialize('PrivateLinkServiceVisibility', pipeline_response) 734 735 if cls: 736 return cls(pipeline_response, deserialized, {}) 737 738 return deserialized 739 _check_private_link_service_visibility_initial.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/locations/{location}/checkPrivateLinkServiceVisibility'} # type: ignore 740 741 def begin_check_private_link_service_visibility( 742 self, 743 location, # type: str 744 parameters, # type: "_models.CheckPrivateLinkServiceVisibilityRequest" 745 **kwargs # type: Any 746 ): 747 # type: (...) -> LROPoller["_models.PrivateLinkServiceVisibility"] 748 """Checks whether the subscription is visible to private link service. 749 750 :param location: The location of the domain name. 751 :type location: str 752 :param parameters: The request body of CheckPrivateLinkService API call. 753 :type parameters: ~azure.mgmt.network.v2019_07_01.models.CheckPrivateLinkServiceVisibilityRequest 754 :keyword callable cls: A custom type or function that will be passed the direct response 755 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 756 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 757 False for no polling, or your own initialized polling object for a personal polling strategy. 758 :paramtype polling: bool or ~azure.core.polling.PollingMethod 759 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 760 :return: An instance of LROPoller that returns either PrivateLinkServiceVisibility or the result of cls(response) 761 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.network.v2019_07_01.models.PrivateLinkServiceVisibility] 762 :raises ~azure.core.exceptions.HttpResponseError: 763 """ 764 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 765 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkServiceVisibility"] 766 lro_delay = kwargs.pop( 767 'polling_interval', 768 self._config.polling_interval 769 ) 770 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 771 if cont_token is None: 772 raw_result = self._check_private_link_service_visibility_initial( 773 location=location, 774 parameters=parameters, 775 cls=lambda x,y,z: x, 776 **kwargs 777 ) 778 779 kwargs.pop('error_map', None) 780 kwargs.pop('content_type', None) 781 782 def get_long_running_output(pipeline_response): 783 deserialized = self._deserialize('PrivateLinkServiceVisibility', pipeline_response) 784 785 if cls: 786 return cls(pipeline_response, deserialized, {}) 787 return deserialized 788 789 path_format_arguments = { 790 'location': self._serialize.url("location", location, 'str'), 791 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 792 } 793 794 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 795 elif polling is False: polling_method = NoPolling() 796 else: polling_method = polling 797 if cont_token: 798 return LROPoller.from_continuation_token( 799 polling_method=polling_method, 800 continuation_token=cont_token, 801 client=self._client, 802 deserialization_callback=get_long_running_output 803 ) 804 else: 805 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 806 begin_check_private_link_service_visibility.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/locations/{location}/checkPrivateLinkServiceVisibility'} # type: ignore 807 808 def _check_private_link_service_visibility_by_resource_group_initial( 809 self, 810 location, # type: str 811 resource_group_name, # type: str 812 parameters, # type: "_models.CheckPrivateLinkServiceVisibilityRequest" 813 **kwargs # type: Any 814 ): 815 # type: (...) -> Optional["_models.PrivateLinkServiceVisibility"] 816 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.PrivateLinkServiceVisibility"]] 817 error_map = { 818 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 819 } 820 error_map.update(kwargs.pop('error_map', {})) 821 api_version = "2019-07-01" 822 content_type = kwargs.pop("content_type", "application/json") 823 accept = "application/json" 824 825 # Construct URL 826 url = self._check_private_link_service_visibility_by_resource_group_initial.metadata['url'] # type: ignore 827 path_format_arguments = { 828 'location': self._serialize.url("location", location, 'str'), 829 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 830 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 831 } 832 url = self._client.format_url(url, **path_format_arguments) 833 834 # Construct parameters 835 query_parameters = {} # type: Dict[str, Any] 836 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 837 838 # Construct headers 839 header_parameters = {} # type: Dict[str, Any] 840 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 841 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 842 843 body_content_kwargs = {} # type: Dict[str, Any] 844 body_content = self._serialize.body(parameters, 'CheckPrivateLinkServiceVisibilityRequest') 845 body_content_kwargs['content'] = body_content 846 request = self._client.post(url, query_parameters, header_parameters, **body_content_kwargs) 847 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 848 response = pipeline_response.http_response 849 850 if response.status_code not in [200, 202]: 851 map_error(status_code=response.status_code, response=response, error_map=error_map) 852 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 853 854 deserialized = None 855 if response.status_code == 200: 856 deserialized = self._deserialize('PrivateLinkServiceVisibility', pipeline_response) 857 858 if cls: 859 return cls(pipeline_response, deserialized, {}) 860 861 return deserialized 862 _check_private_link_service_visibility_by_resource_group_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/locations/{location}/checkPrivateLinkServiceVisibility'} # type: ignore 863 864 def begin_check_private_link_service_visibility_by_resource_group( 865 self, 866 location, # type: str 867 resource_group_name, # type: str 868 parameters, # type: "_models.CheckPrivateLinkServiceVisibilityRequest" 869 **kwargs # type: Any 870 ): 871 # type: (...) -> LROPoller["_models.PrivateLinkServiceVisibility"] 872 """Checks whether the subscription is visible to private link service in the specified resource 873 group. 874 875 :param location: The location of the domain name. 876 :type location: str 877 :param resource_group_name: The name of the resource group. 878 :type resource_group_name: str 879 :param parameters: The request body of CheckPrivateLinkService API call. 880 :type parameters: ~azure.mgmt.network.v2019_07_01.models.CheckPrivateLinkServiceVisibilityRequest 881 :keyword callable cls: A custom type or function that will be passed the direct response 882 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 883 :keyword polling: Pass in True if you'd like the ARMPolling polling method, 884 False for no polling, or your own initialized polling object for a personal polling strategy. 885 :paramtype polling: bool or ~azure.core.polling.PollingMethod 886 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 887 :return: An instance of LROPoller that returns either PrivateLinkServiceVisibility or the result of cls(response) 888 :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.network.v2019_07_01.models.PrivateLinkServiceVisibility] 889 :raises ~azure.core.exceptions.HttpResponseError: 890 """ 891 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 892 cls = kwargs.pop('cls', None) # type: ClsType["_models.PrivateLinkServiceVisibility"] 893 lro_delay = kwargs.pop( 894 'polling_interval', 895 self._config.polling_interval 896 ) 897 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 898 if cont_token is None: 899 raw_result = self._check_private_link_service_visibility_by_resource_group_initial( 900 location=location, 901 resource_group_name=resource_group_name, 902 parameters=parameters, 903 cls=lambda x,y,z: x, 904 **kwargs 905 ) 906 907 kwargs.pop('error_map', None) 908 kwargs.pop('content_type', None) 909 910 def get_long_running_output(pipeline_response): 911 deserialized = self._deserialize('PrivateLinkServiceVisibility', pipeline_response) 912 913 if cls: 914 return cls(pipeline_response, deserialized, {}) 915 return deserialized 916 917 path_format_arguments = { 918 'location': self._serialize.url("location", location, 'str'), 919 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 920 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 921 } 922 923 if polling is True: polling_method = ARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 924 elif polling is False: polling_method = NoPolling() 925 else: polling_method = polling 926 if cont_token: 927 return LROPoller.from_continuation_token( 928 polling_method=polling_method, 929 continuation_token=cont_token, 930 client=self._client, 931 deserialization_callback=get_long_running_output 932 ) 933 else: 934 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 935 begin_check_private_link_service_visibility_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/locations/{location}/checkPrivateLinkServiceVisibility'} # type: ignore 936 937 def list_auto_approved_private_link_services( 938 self, 939 location, # type: str 940 **kwargs # type: Any 941 ): 942 # type: (...) -> Iterable["_models.AutoApprovedPrivateLinkServicesResult"] 943 """Returns all of the private link service ids that can be linked to a Private Endpoint with auto 944 approved in this subscription in this region. 945 946 :param location: The location of the domain name. 947 :type location: str 948 :keyword callable cls: A custom type or function that will be passed the direct response 949 :return: An iterator like instance of either AutoApprovedPrivateLinkServicesResult or the result of cls(response) 950 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_07_01.models.AutoApprovedPrivateLinkServicesResult] 951 :raises: ~azure.core.exceptions.HttpResponseError 952 """ 953 cls = kwargs.pop('cls', None) # type: ClsType["_models.AutoApprovedPrivateLinkServicesResult"] 954 error_map = { 955 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 956 } 957 error_map.update(kwargs.pop('error_map', {})) 958 api_version = "2019-07-01" 959 accept = "application/json" 960 961 def prepare_request(next_link=None): 962 # Construct headers 963 header_parameters = {} # type: Dict[str, Any] 964 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 965 966 if not next_link: 967 # Construct URL 968 url = self.list_auto_approved_private_link_services.metadata['url'] # type: ignore 969 path_format_arguments = { 970 'location': self._serialize.url("location", location, 'str'), 971 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 972 } 973 url = self._client.format_url(url, **path_format_arguments) 974 # Construct parameters 975 query_parameters = {} # type: Dict[str, Any] 976 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 977 978 request = self._client.get(url, query_parameters, header_parameters) 979 else: 980 url = next_link 981 query_parameters = {} # type: Dict[str, Any] 982 request = self._client.get(url, query_parameters, header_parameters) 983 return request 984 985 def extract_data(pipeline_response): 986 deserialized = self._deserialize('AutoApprovedPrivateLinkServicesResult', pipeline_response) 987 list_of_elem = deserialized.value 988 if cls: 989 list_of_elem = cls(list_of_elem) 990 return deserialized.next_link or None, iter(list_of_elem) 991 992 def get_next(next_link=None): 993 request = prepare_request(next_link) 994 995 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 996 response = pipeline_response.http_response 997 998 if response.status_code not in [200]: 999 map_error(status_code=response.status_code, response=response, error_map=error_map) 1000 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1001 1002 return pipeline_response 1003 1004 return ItemPaged( 1005 get_next, extract_data 1006 ) 1007 list_auto_approved_private_link_services.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/locations/{location}/autoApprovedPrivateLinkServices'} # type: ignore 1008 1009 def list_auto_approved_private_link_services_by_resource_group( 1010 self, 1011 location, # type: str 1012 resource_group_name, # type: str 1013 **kwargs # type: Any 1014 ): 1015 # type: (...) -> Iterable["_models.AutoApprovedPrivateLinkServicesResult"] 1016 """Returns all of the private link service ids that can be linked to a Private Endpoint with auto 1017 approved in this subscription in this region. 1018 1019 :param location: The location of the domain name. 1020 :type location: str 1021 :param resource_group_name: The name of the resource group. 1022 :type resource_group_name: str 1023 :keyword callable cls: A custom type or function that will be passed the direct response 1024 :return: An iterator like instance of either AutoApprovedPrivateLinkServicesResult or the result of cls(response) 1025 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.network.v2019_07_01.models.AutoApprovedPrivateLinkServicesResult] 1026 :raises: ~azure.core.exceptions.HttpResponseError 1027 """ 1028 cls = kwargs.pop('cls', None) # type: ClsType["_models.AutoApprovedPrivateLinkServicesResult"] 1029 error_map = { 1030 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 1031 } 1032 error_map.update(kwargs.pop('error_map', {})) 1033 api_version = "2019-07-01" 1034 accept = "application/json" 1035 1036 def prepare_request(next_link=None): 1037 # Construct headers 1038 header_parameters = {} # type: Dict[str, Any] 1039 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 1040 1041 if not next_link: 1042 # Construct URL 1043 url = self.list_auto_approved_private_link_services_by_resource_group.metadata['url'] # type: ignore 1044 path_format_arguments = { 1045 'location': self._serialize.url("location", location, 'str'), 1046 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 1047 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 1048 } 1049 url = self._client.format_url(url, **path_format_arguments) 1050 # Construct parameters 1051 query_parameters = {} # type: Dict[str, Any] 1052 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 1053 1054 request = self._client.get(url, query_parameters, header_parameters) 1055 else: 1056 url = next_link 1057 query_parameters = {} # type: Dict[str, Any] 1058 request = self._client.get(url, query_parameters, header_parameters) 1059 return request 1060 1061 def extract_data(pipeline_response): 1062 deserialized = self._deserialize('AutoApprovedPrivateLinkServicesResult', pipeline_response) 1063 list_of_elem = deserialized.value 1064 if cls: 1065 list_of_elem = cls(list_of_elem) 1066 return deserialized.next_link or None, iter(list_of_elem) 1067 1068 def get_next(next_link=None): 1069 request = prepare_request(next_link) 1070 1071 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 1072 response = pipeline_response.http_response 1073 1074 if response.status_code not in [200]: 1075 map_error(status_code=response.status_code, response=response, error_map=error_map) 1076 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 1077 1078 return pipeline_response 1079 1080 return ItemPaged( 1081 get_next, extract_data 1082 ) 1083 list_auto_approved_private_link_services_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/locations/{location}/autoApprovedPrivateLinkServices'} # type: ignore 1084