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 AzureFirewallsOperations: 25 """AzureFirewallsOperations 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_03_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 azure_firewall_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-03-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 'azureFirewallName': self._serialize.url("azure_firewall_name", azure_firewall_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/azureFirewalls/{azureFirewallName}'} # type: ignore 89 90 async def begin_delete( 91 self, 92 resource_group_name: str, 93 azure_firewall_name: str, 94 **kwargs 95 ) -> AsyncLROPoller[None]: 96 """Deletes the specified Azure Firewall. 97 98 :param resource_group_name: The name of the resource group. 99 :type resource_group_name: str 100 :param azure_firewall_name: The name of the Azure Firewall. 101 :type azure_firewall_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 azure_firewall_name=azure_firewall_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 'azureFirewallName': self._serialize.url("azure_firewall_name", azure_firewall_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/azureFirewalls/{azureFirewallName}'} # type: ignore 153 154 async def get( 155 self, 156 resource_group_name: str, 157 azure_firewall_name: str, 158 **kwargs 159 ) -> "_models.AzureFirewall": 160 """Gets the specified Azure Firewall. 161 162 :param resource_group_name: The name of the resource group. 163 :type resource_group_name: str 164 :param azure_firewall_name: The name of the Azure Firewall. 165 :type azure_firewall_name: str 166 :keyword callable cls: A custom type or function that will be passed the direct response 167 :return: AzureFirewall, or the result of cls(response) 168 :rtype: ~azure.mgmt.network.v2020_03_01.models.AzureFirewall 169 :raises: ~azure.core.exceptions.HttpResponseError 170 """ 171 cls = kwargs.pop('cls', None) # type: ClsType["_models.AzureFirewall"] 172 error_map = { 173 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 174 } 175 error_map.update(kwargs.pop('error_map', {})) 176 api_version = "2020-03-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 'azureFirewallName': self._serialize.url("azure_firewall_name", azure_firewall_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('AzureFirewall', 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/azureFirewalls/{azureFirewallName}'} # type: ignore 211 212 async def _create_or_update_initial( 213 self, 214 resource_group_name: str, 215 azure_firewall_name: str, 216 parameters: "_models.AzureFirewall", 217 **kwargs 218 ) -> "_models.AzureFirewall": 219 cls = kwargs.pop('cls', None) # type: ClsType["_models.AzureFirewall"] 220 error_map = { 221 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 222 } 223 error_map.update(kwargs.pop('error_map', {})) 224 api_version = "2020-03-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 'azureFirewallName': self._serialize.url("azure_firewall_name", azure_firewall_name, 'str', max_length=56, min_length=1), 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, 'AzureFirewall') 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('AzureFirewall', pipeline_response) 259 260 if response.status_code == 201: 261 deserialized = self._deserialize('AzureFirewall', 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/azureFirewalls/{azureFirewallName}'} # type: ignore 268 269 async def begin_create_or_update( 270 self, 271 resource_group_name: str, 272 azure_firewall_name: str, 273 parameters: "_models.AzureFirewall", 274 **kwargs 275 ) -> AsyncLROPoller["_models.AzureFirewall"]: 276 """Creates or updates the specified Azure Firewall. 277 278 :param resource_group_name: The name of the resource group. 279 :type resource_group_name: str 280 :param azure_firewall_name: The name of the Azure Firewall. 281 :type azure_firewall_name: str 282 :param parameters: Parameters supplied to the create or update Azure Firewall operation. 283 :type parameters: ~azure.mgmt.network.v2020_03_01.models.AzureFirewall 284 :keyword callable cls: A custom type or function that will be passed the direct response 285 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 286 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 287 False for no polling, or your own initialized polling object for a personal polling strategy. 288 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 289 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 290 :return: An instance of AsyncLROPoller that returns either AzureFirewall or the result of cls(response) 291 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_03_01.models.AzureFirewall] 292 :raises ~azure.core.exceptions.HttpResponseError: 293 """ 294 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 295 cls = kwargs.pop('cls', None) # type: ClsType["_models.AzureFirewall"] 296 lro_delay = kwargs.pop( 297 'polling_interval', 298 self._config.polling_interval 299 ) 300 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 301 if cont_token is None: 302 raw_result = await self._create_or_update_initial( 303 resource_group_name=resource_group_name, 304 azure_firewall_name=azure_firewall_name, 305 parameters=parameters, 306 cls=lambda x,y,z: x, 307 **kwargs 308 ) 309 310 kwargs.pop('error_map', None) 311 kwargs.pop('content_type', None) 312 313 def get_long_running_output(pipeline_response): 314 deserialized = self._deserialize('AzureFirewall', pipeline_response) 315 316 if cls: 317 return cls(pipeline_response, deserialized, {}) 318 return deserialized 319 320 path_format_arguments = { 321 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 322 'azureFirewallName': self._serialize.url("azure_firewall_name", azure_firewall_name, 'str', max_length=56, min_length=1), 323 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 324 } 325 326 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 327 elif polling is False: polling_method = AsyncNoPolling() 328 else: polling_method = polling 329 if cont_token: 330 return AsyncLROPoller.from_continuation_token( 331 polling_method=polling_method, 332 continuation_token=cont_token, 333 client=self._client, 334 deserialization_callback=get_long_running_output 335 ) 336 else: 337 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 338 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/azureFirewalls/{azureFirewallName}'} # type: ignore 339 340 async def _update_tags_initial( 341 self, 342 resource_group_name: str, 343 azure_firewall_name: str, 344 parameters: "_models.TagsObject", 345 **kwargs 346 ) -> Optional["_models.AzureFirewall"]: 347 cls = kwargs.pop('cls', None) # type: ClsType[Optional["_models.AzureFirewall"]] 348 error_map = { 349 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 350 } 351 error_map.update(kwargs.pop('error_map', {})) 352 api_version = "2020-03-01" 353 content_type = kwargs.pop("content_type", "application/json") 354 accept = "application/json" 355 356 # Construct URL 357 url = self._update_tags_initial.metadata['url'] # type: ignore 358 path_format_arguments = { 359 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 360 'azureFirewallName': self._serialize.url("azure_firewall_name", azure_firewall_name, 'str'), 361 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 362 } 363 url = self._client.format_url(url, **path_format_arguments) 364 365 # Construct parameters 366 query_parameters = {} # type: Dict[str, Any] 367 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 368 369 # Construct headers 370 header_parameters = {} # type: Dict[str, Any] 371 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 372 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 373 374 body_content_kwargs = {} # type: Dict[str, Any] 375 body_content = self._serialize.body(parameters, 'TagsObject') 376 body_content_kwargs['content'] = body_content 377 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 378 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 379 response = pipeline_response.http_response 380 381 if response.status_code not in [200, 202]: 382 map_error(status_code=response.status_code, response=response, error_map=error_map) 383 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 384 385 deserialized = None 386 if response.status_code == 200: 387 deserialized = self._deserialize('AzureFirewall', pipeline_response) 388 389 if cls: 390 return cls(pipeline_response, deserialized, {}) 391 392 return deserialized 393 _update_tags_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/azureFirewalls/{azureFirewallName}'} # type: ignore 394 395 async def begin_update_tags( 396 self, 397 resource_group_name: str, 398 azure_firewall_name: str, 399 parameters: "_models.TagsObject", 400 **kwargs 401 ) -> AsyncLROPoller["_models.AzureFirewall"]: 402 """Updates tags of an Azure Firewall resource. 403 404 :param resource_group_name: The name of the resource group. 405 :type resource_group_name: str 406 :param azure_firewall_name: The name of the Azure Firewall. 407 :type azure_firewall_name: str 408 :param parameters: Parameters supplied to update azure firewall tags. 409 :type parameters: ~azure.mgmt.network.v2020_03_01.models.TagsObject 410 :keyword callable cls: A custom type or function that will be passed the direct response 411 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 412 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 413 False for no polling, or your own initialized polling object for a personal polling strategy. 414 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 415 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 416 :return: An instance of AsyncLROPoller that returns either AzureFirewall or the result of cls(response) 417 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2020_03_01.models.AzureFirewall] 418 :raises ~azure.core.exceptions.HttpResponseError: 419 """ 420 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 421 cls = kwargs.pop('cls', None) # type: ClsType["_models.AzureFirewall"] 422 lro_delay = kwargs.pop( 423 'polling_interval', 424 self._config.polling_interval 425 ) 426 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 427 if cont_token is None: 428 raw_result = await self._update_tags_initial( 429 resource_group_name=resource_group_name, 430 azure_firewall_name=azure_firewall_name, 431 parameters=parameters, 432 cls=lambda x,y,z: x, 433 **kwargs 434 ) 435 436 kwargs.pop('error_map', None) 437 kwargs.pop('content_type', None) 438 439 def get_long_running_output(pipeline_response): 440 deserialized = self._deserialize('AzureFirewall', pipeline_response) 441 442 if cls: 443 return cls(pipeline_response, deserialized, {}) 444 return deserialized 445 446 path_format_arguments = { 447 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 448 'azureFirewallName': self._serialize.url("azure_firewall_name", azure_firewall_name, 'str'), 449 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 450 } 451 452 if polling is True: polling_method = AsyncARMPolling(lro_delay, lro_options={'final-state-via': 'azure-async-operation'}, path_format_arguments=path_format_arguments, **kwargs) 453 elif polling is False: polling_method = AsyncNoPolling() 454 else: polling_method = polling 455 if cont_token: 456 return AsyncLROPoller.from_continuation_token( 457 polling_method=polling_method, 458 continuation_token=cont_token, 459 client=self._client, 460 deserialization_callback=get_long_running_output 461 ) 462 else: 463 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 464 begin_update_tags.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/azureFirewalls/{azureFirewallName}'} # type: ignore 465 466 def list( 467 self, 468 resource_group_name: str, 469 **kwargs 470 ) -> AsyncIterable["_models.AzureFirewallListResult"]: 471 """Lists all Azure Firewalls in a resource group. 472 473 :param resource_group_name: The name of the resource group. 474 :type resource_group_name: str 475 :keyword callable cls: A custom type or function that will be passed the direct response 476 :return: An iterator like instance of either AzureFirewallListResult or the result of cls(response) 477 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_03_01.models.AzureFirewallListResult] 478 :raises: ~azure.core.exceptions.HttpResponseError 479 """ 480 cls = kwargs.pop('cls', None) # type: ClsType["_models.AzureFirewallListResult"] 481 error_map = { 482 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 483 } 484 error_map.update(kwargs.pop('error_map', {})) 485 api_version = "2020-03-01" 486 accept = "application/json" 487 488 def prepare_request(next_link=None): 489 # Construct headers 490 header_parameters = {} # type: Dict[str, Any] 491 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 492 493 if not next_link: 494 # Construct URL 495 url = self.list.metadata['url'] # type: ignore 496 path_format_arguments = { 497 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 498 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 499 } 500 url = self._client.format_url(url, **path_format_arguments) 501 # Construct parameters 502 query_parameters = {} # type: Dict[str, Any] 503 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 504 505 request = self._client.get(url, query_parameters, header_parameters) 506 else: 507 url = next_link 508 query_parameters = {} # type: Dict[str, Any] 509 request = self._client.get(url, query_parameters, header_parameters) 510 return request 511 512 async def extract_data(pipeline_response): 513 deserialized = self._deserialize('AzureFirewallListResult', pipeline_response) 514 list_of_elem = deserialized.value 515 if cls: 516 list_of_elem = cls(list_of_elem) 517 return deserialized.next_link or None, AsyncList(list_of_elem) 518 519 async def get_next(next_link=None): 520 request = prepare_request(next_link) 521 522 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 523 response = pipeline_response.http_response 524 525 if response.status_code not in [200]: 526 map_error(status_code=response.status_code, response=response, error_map=error_map) 527 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 528 529 return pipeline_response 530 531 return AsyncItemPaged( 532 get_next, extract_data 533 ) 534 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/azureFirewalls'} # type: ignore 535 536 def list_all( 537 self, 538 **kwargs 539 ) -> AsyncIterable["_models.AzureFirewallListResult"]: 540 """Gets all the Azure Firewalls in a subscription. 541 542 :keyword callable cls: A custom type or function that will be passed the direct response 543 :return: An iterator like instance of either AzureFirewallListResult or the result of cls(response) 544 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2020_03_01.models.AzureFirewallListResult] 545 :raises: ~azure.core.exceptions.HttpResponseError 546 """ 547 cls = kwargs.pop('cls', None) # type: ClsType["_models.AzureFirewallListResult"] 548 error_map = { 549 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 550 } 551 error_map.update(kwargs.pop('error_map', {})) 552 api_version = "2020-03-01" 553 accept = "application/json" 554 555 def prepare_request(next_link=None): 556 # Construct headers 557 header_parameters = {} # type: Dict[str, Any] 558 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 559 560 if not next_link: 561 # Construct URL 562 url = self.list_all.metadata['url'] # type: ignore 563 path_format_arguments = { 564 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 565 } 566 url = self._client.format_url(url, **path_format_arguments) 567 # Construct parameters 568 query_parameters = {} # type: Dict[str, Any] 569 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 570 571 request = self._client.get(url, query_parameters, header_parameters) 572 else: 573 url = next_link 574 query_parameters = {} # type: Dict[str, Any] 575 request = self._client.get(url, query_parameters, header_parameters) 576 return request 577 578 async def extract_data(pipeline_response): 579 deserialized = self._deserialize('AzureFirewallListResult', pipeline_response) 580 list_of_elem = deserialized.value 581 if cls: 582 list_of_elem = cls(list_of_elem) 583 return deserialized.next_link or None, AsyncList(list_of_elem) 584 585 async def get_next(next_link=None): 586 request = prepare_request(next_link) 587 588 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 589 response = pipeline_response.http_response 590 591 if response.status_code not in [200]: 592 map_error(status_code=response.status_code, response=response, error_map=error_map) 593 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 594 595 return pipeline_response 596 597 return AsyncItemPaged( 598 get_next, extract_data 599 ) 600 list_all.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/azureFirewalls'} # type: ignore 601