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