1# coding=utf-8 2# -------------------------------------------------------------------------- 3# Copyright (c) Microsoft Corporation. All rights reserved. 4# Licensed under the MIT License. See License.txt in the project root for license information. 5# Code generated by Microsoft (R) AutoRest Code Generator. 6# Changes may cause incorrect behavior and will be lost if the code is regenerated. 7# -------------------------------------------------------------------------- 8from typing import Any, AsyncIterable, Callable, Dict, Generic, Optional, TypeVar, Union 9import warnings 10 11from azure.core.async_paging import AsyncItemPaged, AsyncList 12from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error 13from azure.core.pipeline import PipelineResponse 14from azure.core.pipeline.transport import AsyncHttpResponse, HttpRequest 15from azure.core.polling import AsyncLROPoller, AsyncNoPolling, AsyncPollingMethod 16from azure.mgmt.core.exceptions import ARMErrorFormat 17from azure.mgmt.core.polling.async_arm_polling import AsyncARMPolling 18 19from ... import models as _models 20 21T = TypeVar('T') 22ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]] 23 24class ApplicationSecurityGroupsOperations: 25 """ApplicationSecurityGroupsOperations async operations. 26 27 You should not instantiate this class directly. Instead, you should create a Client instance that 28 instantiates it for you and attaches it as an attribute. 29 30 :ivar models: Alias to model classes used in this operation group. 31 :type models: ~azure.mgmt.network.v2020_06_01.models 32 :param client: Client for service requests. 33 :param config: Configuration of service client. 34 :param serializer: An object model serializer. 35 :param deserializer: An object model deserializer. 36 """ 37 38 models = _models 39 40 def __init__(self, client, config, serializer, deserializer) -> None: 41 self._client = client 42 self._serialize = serializer 43 self._deserialize = deserializer 44 self._config = config 45 46 async def _delete_initial( 47 self, 48 resource_group_name: str, 49 application_security_group_name: str, 50 **kwargs 51 ) -> None: 52 cls = kwargs.pop('cls', None) # type: ClsType[None] 53 error_map = { 54 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 55 } 56 error_map.update(kwargs.pop('error_map', {})) 57 api_version = "2020-06-01" 58 accept = "application/json" 59 60 # Construct URL 61 url = self._delete_initial.metadata['url'] # type: ignore 62 path_format_arguments = { 63 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 64 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 65 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 66 } 67 url = self._client.format_url(url, **path_format_arguments) 68 69 # Construct parameters 70 query_parameters = {} # type: Dict[str, Any] 71 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 72 73 # Construct headers 74 header_parameters = {} # type: Dict[str, Any] 75 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 76 77 request = self._client.delete(url, query_parameters, header_parameters) 78 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 79 response = pipeline_response.http_response 80 81 if response.status_code not in [200, 202, 204]: 82 map_error(status_code=response.status_code, response=response, error_map=error_map) 83 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 84 85 if cls: 86 return cls(pipeline_response, None, {}) 87 88 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 89 90 async def begin_delete( 91 self, 92 resource_group_name: str, 93 application_security_group_name: str, 94 **kwargs 95 ) -> AsyncLROPoller[None]: 96 """Deletes the specified application security group. 97 98 :param resource_group_name: The name of the resource group. 99 :type resource_group_name: str 100 :param application_security_group_name: The name of the application security group. 101 :type application_security_group_name: str 102 :keyword callable cls: A custom type or function that will be passed the direct response 103 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 104 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 105 False for no polling, or your own initialized polling object for a personal polling strategy. 106 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 107 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 108 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 109 :rtype: ~azure.core.polling.AsyncLROPoller[None] 110 :raises ~azure.core.exceptions.HttpResponseError: 111 """ 112 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 113 cls = kwargs.pop('cls', None) # type: ClsType[None] 114 lro_delay = kwargs.pop( 115 'polling_interval', 116 self._config.polling_interval 117 ) 118 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 119 if cont_token is None: 120 raw_result = await self._delete_initial( 121 resource_group_name=resource_group_name, 122 application_security_group_name=application_security_group_name, 123 cls=lambda x,y,z: x, 124 **kwargs 125 ) 126 127 kwargs.pop('error_map', None) 128 kwargs.pop('content_type', None) 129 130 def get_long_running_output(pipeline_response): 131 if cls: 132 return cls(pipeline_response, None, {}) 133 134 path_format_arguments = { 135 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 136 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 137 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 138 } 139 140 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) 141 elif polling is False: polling_method = AsyncNoPolling() 142 else: polling_method = polling 143 if cont_token: 144 return AsyncLROPoller.from_continuation_token( 145 polling_method=polling_method, 146 continuation_token=cont_token, 147 client=self._client, 148 deserialization_callback=get_long_running_output 149 ) 150 else: 151 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 152 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 153 154 async def get( 155 self, 156 resource_group_name: str, 157 application_security_group_name: str, 158 **kwargs 159 ) -> "_models.ApplicationSecurityGroup": 160 """Gets information about the specified application security group. 161 162 :param resource_group_name: The name of the resource group. 163 :type resource_group_name: str 164 :param application_security_group_name: The name of the application security group. 165 :type application_security_group_name: str 166 :keyword callable cls: A custom type or function that will be passed the direct response 167 :return: ApplicationSecurityGroup, or the result of cls(response) 168 :rtype: ~azure.mgmt.network.v2020_06_01.models.ApplicationSecurityGroup 169 :raises: ~azure.core.exceptions.HttpResponseError 170 """ 171 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroup"] 172 error_map = { 173 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 174 } 175 error_map.update(kwargs.pop('error_map', {})) 176 api_version = "2020-06-01" 177 accept = "application/json" 178 179 # Construct URL 180 url = self.get.metadata['url'] # type: ignore 181 path_format_arguments = { 182 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 183 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 184 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 185 } 186 url = self._client.format_url(url, **path_format_arguments) 187 188 # Construct parameters 189 query_parameters = {} # type: Dict[str, Any] 190 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 191 192 # Construct headers 193 header_parameters = {} # type: Dict[str, Any] 194 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 195 196 request = self._client.get(url, query_parameters, header_parameters) 197 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 198 response = pipeline_response.http_response 199 200 if response.status_code not in [200]: 201 map_error(status_code=response.status_code, response=response, error_map=error_map) 202 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 203 204 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 205 206 if cls: 207 return cls(pipeline_response, deserialized, {}) 208 209 return deserialized 210 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 211 212 async def _create_or_update_initial( 213 self, 214 resource_group_name: str, 215 application_security_group_name: str, 216 parameters: "_models.ApplicationSecurityGroup", 217 **kwargs 218 ) -> "_models.ApplicationSecurityGroup": 219 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroup"] 220 error_map = { 221 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 222 } 223 error_map.update(kwargs.pop('error_map', {})) 224 api_version = "2020-06-01" 225 content_type = kwargs.pop("content_type", "application/json") 226 accept = "application/json" 227 228 # Construct URL 229 url = self._create_or_update_initial.metadata['url'] # type: ignore 230 path_format_arguments = { 231 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 232 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 233 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 234 } 235 url = self._client.format_url(url, **path_format_arguments) 236 237 # Construct parameters 238 query_parameters = {} # type: Dict[str, Any] 239 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 240 241 # Construct headers 242 header_parameters = {} # type: Dict[str, Any] 243 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 244 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 245 246 body_content_kwargs = {} # type: Dict[str, Any] 247 body_content = self._serialize.body(parameters, 'ApplicationSecurityGroup') 248 body_content_kwargs['content'] = body_content 249 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 250 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 251 response = pipeline_response.http_response 252 253 if response.status_code not in [200, 201]: 254 map_error(status_code=response.status_code, response=response, error_map=error_map) 255 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 256 257 if response.status_code == 200: 258 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 259 260 if response.status_code == 201: 261 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 262 263 if cls: 264 return cls(pipeline_response, deserialized, {}) 265 266 return deserialized 267 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 268 269 async def begin_create_or_update( 270 self, 271 resource_group_name: str, 272 application_security_group_name: str, 273 parameters: "_models.ApplicationSecurityGroup", 274 **kwargs 275 ) -> AsyncLROPoller["_models.ApplicationSecurityGroup"]: 276 """Creates or updates an application security group. 277 278 :param resource_group_name: The name of the resource group. 279 :type resource_group_name: str 280 :param application_security_group_name: The name of the application security group. 281 :type application_security_group_name: str 282 :param parameters: Parameters supplied to the create or update ApplicationSecurityGroup 283 operation. 284 :type parameters: ~azure.mgmt.network.v2020_06_01.models.ApplicationSecurityGroup 285 :keyword callable cls: A custom type or function that will be passed the direct response 286 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 287 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 288 False for no polling, or your own initialized polling object for a personal polling strategy. 289 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 290 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 291 :return: An instance of AsyncLROPoller that returns either ApplicationSecurityGroup or the result of cls(response) 292 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_06_01.models.ApplicationSecurityGroup] 293 :raises ~azure.core.exceptions.HttpResponseError: 294 """ 295 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 296 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroup"] 297 lro_delay = kwargs.pop( 298 'polling_interval', 299 self._config.polling_interval 300 ) 301 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 302 if cont_token is None: 303 raw_result = await self._create_or_update_initial( 304 resource_group_name=resource_group_name, 305 application_security_group_name=application_security_group_name, 306 parameters=parameters, 307 cls=lambda x,y,z: x, 308 **kwargs 309 ) 310 311 kwargs.pop('error_map', None) 312 kwargs.pop('content_type', None) 313 314 def get_long_running_output(pipeline_response): 315 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 316 317 if cls: 318 return cls(pipeline_response, deserialized, {}) 319 return deserialized 320 321 path_format_arguments = { 322 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 323 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 324 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 325 } 326 327 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 328 elif polling is False: polling_method = AsyncNoPolling() 329 else: polling_method = polling 330 if cont_token: 331 return AsyncLROPoller.from_continuation_token( 332 polling_method=polling_method, 333 continuation_token=cont_token, 334 client=self._client, 335 deserialization_callback=get_long_running_output 336 ) 337 else: 338 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 339 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 340 341 async def update_tags( 342 self, 343 resource_group_name: str, 344 application_security_group_name: str, 345 parameters: "_models.TagsObject", 346 **kwargs 347 ) -> "_models.ApplicationSecurityGroup": 348 """Updates an application security group's tags. 349 350 :param resource_group_name: The name of the resource group. 351 :type resource_group_name: str 352 :param application_security_group_name: The name of the application security group. 353 :type application_security_group_name: str 354 :param parameters: Parameters supplied to update application security group tags. 355 :type parameters: ~azure.mgmt.network.v2020_06_01.models.TagsObject 356 :keyword callable cls: A custom type or function that will be passed the direct response 357 :return: ApplicationSecurityGroup, or the result of cls(response) 358 :rtype: ~azure.mgmt.network.v2020_06_01.models.ApplicationSecurityGroup 359 :raises: ~azure.core.exceptions.HttpResponseError 360 """ 361 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroup"] 362 error_map = { 363 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 364 } 365 error_map.update(kwargs.pop('error_map', {})) 366 api_version = "2020-06-01" 367 content_type = kwargs.pop("content_type", "application/json") 368 accept = "application/json" 369 370 # Construct URL 371 url = self.update_tags.metadata['url'] # type: ignore 372 path_format_arguments = { 373 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 374 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 375 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 376 } 377 url = self._client.format_url(url, **path_format_arguments) 378 379 # Construct parameters 380 query_parameters = {} # type: Dict[str, Any] 381 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 382 383 # Construct headers 384 header_parameters = {} # type: Dict[str, Any] 385 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 386 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 387 388 body_content_kwargs = {} # type: Dict[str, Any] 389 body_content = self._serialize.body(parameters, 'TagsObject') 390 body_content_kwargs['content'] = body_content 391 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 392 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 393 response = pipeline_response.http_response 394 395 if response.status_code not in [200]: 396 map_error(status_code=response.status_code, response=response, error_map=error_map) 397 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 398 399 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 400 401 if cls: 402 return cls(pipeline_response, deserialized, {}) 403 404 return deserialized 405 update_tags.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 406 407 def list_all( 408 self, 409 **kwargs 410 ) -> AsyncIterable["_models.ApplicationSecurityGroupListResult"]: 411 """Gets all application security groups in a subscription. 412 413 :keyword callable cls: A custom type or function that will be passed the direct response 414 :return: An iterator like instance of either ApplicationSecurityGroupListResult or the result of cls(response) 415 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_06_01.models.ApplicationSecurityGroupListResult] 416 :raises: ~azure.core.exceptions.HttpResponseError 417 """ 418 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroupListResult"] 419 error_map = { 420 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 421 } 422 error_map.update(kwargs.pop('error_map', {})) 423 api_version = "2020-06-01" 424 accept = "application/json" 425 426 def prepare_request(next_link=None): 427 # Construct headers 428 header_parameters = {} # type: Dict[str, Any] 429 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 430 431 if not next_link: 432 # Construct URL 433 url = self.list_all.metadata['url'] # type: ignore 434 path_format_arguments = { 435 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 436 } 437 url = self._client.format_url(url, **path_format_arguments) 438 # Construct parameters 439 query_parameters = {} # type: Dict[str, Any] 440 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 441 442 request = self._client.get(url, query_parameters, header_parameters) 443 else: 444 url = next_link 445 query_parameters = {} # type: Dict[str, Any] 446 request = self._client.get(url, query_parameters, header_parameters) 447 return request 448 449 async def extract_data(pipeline_response): 450 deserialized = self._deserialize('ApplicationSecurityGroupListResult', pipeline_response) 451 list_of_elem = deserialized.value 452 if cls: 453 list_of_elem = cls(list_of_elem) 454 return deserialized.next_link or None, AsyncList(list_of_elem) 455 456 async def get_next(next_link=None): 457 request = prepare_request(next_link) 458 459 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 460 response = pipeline_response.http_response 461 462 if response.status_code not in [200]: 463 map_error(status_code=response.status_code, response=response, error_map=error_map) 464 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 465 466 return pipeline_response 467 468 return AsyncItemPaged( 469 get_next, extract_data 470 ) 471 list_all.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/applicationSecurityGroups'} # type: ignore 472 473 def list( 474 self, 475 resource_group_name: str, 476 **kwargs 477 ) -> AsyncIterable["_models.ApplicationSecurityGroupListResult"]: 478 """Gets all the application security groups in a resource group. 479 480 :param resource_group_name: The name of the resource group. 481 :type resource_group_name: str 482 :keyword callable cls: A custom type or function that will be passed the direct response 483 :return: An iterator like instance of either ApplicationSecurityGroupListResult or the result of cls(response) 484 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_06_01.models.ApplicationSecurityGroupListResult] 485 :raises: ~azure.core.exceptions.HttpResponseError 486 """ 487 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroupListResult"] 488 error_map = { 489 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 490 } 491 error_map.update(kwargs.pop('error_map', {})) 492 api_version = "2020-06-01" 493 accept = "application/json" 494 495 def prepare_request(next_link=None): 496 # Construct headers 497 header_parameters = {} # type: Dict[str, Any] 498 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 499 500 if not next_link: 501 # Construct URL 502 url = self.list.metadata['url'] # type: ignore 503 path_format_arguments = { 504 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 505 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 506 } 507 url = self._client.format_url(url, **path_format_arguments) 508 # Construct parameters 509 query_parameters = {} # type: Dict[str, Any] 510 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 511 512 request = self._client.get(url, query_parameters, header_parameters) 513 else: 514 url = next_link 515 query_parameters = {} # type: Dict[str, Any] 516 request = self._client.get(url, query_parameters, header_parameters) 517 return request 518 519 async def extract_data(pipeline_response): 520 deserialized = self._deserialize('ApplicationSecurityGroupListResult', pipeline_response) 521 list_of_elem = deserialized.value 522 if cls: 523 list_of_elem = cls(list_of_elem) 524 return deserialized.next_link or None, AsyncList(list_of_elem) 525 526 async def get_next(next_link=None): 527 request = prepare_request(next_link) 528 529 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 530 response = pipeline_response.http_response 531 532 if response.status_code not in [200]: 533 map_error(status_code=response.status_code, response=response, error_map=error_map) 534 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 535 536 return pipeline_response 537 538 return AsyncItemPaged( 539 get_next, extract_data 540 ) 541 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups'} # type: ignore 542