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.v2018_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 = "2018-06-01" 58 59 # Construct URL 60 url = self._delete_initial.metadata['url'] # type: ignore 61 path_format_arguments = { 62 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 63 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 64 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 65 } 66 url = self._client.format_url(url, **path_format_arguments) 67 68 # Construct parameters 69 query_parameters = {} # type: Dict[str, Any] 70 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 71 72 # Construct headers 73 header_parameters = {} # type: Dict[str, Any] 74 75 request = self._client.delete(url, query_parameters, header_parameters) 76 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 77 response = pipeline_response.http_response 78 79 if response.status_code not in [200, 202, 204]: 80 map_error(status_code=response.status_code, response=response, error_map=error_map) 81 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 82 83 if cls: 84 return cls(pipeline_response, None, {}) 85 86 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 87 88 async def begin_delete( 89 self, 90 resource_group_name: str, 91 application_security_group_name: str, 92 **kwargs 93 ) -> AsyncLROPoller[None]: 94 """Deletes the specified application security group. 95 96 :param resource_group_name: The name of the resource group. 97 :type resource_group_name: str 98 :param application_security_group_name: The name of the application security group. 99 :type application_security_group_name: str 100 :keyword callable cls: A custom type or function that will be passed the direct response 101 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 102 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 103 False for no polling, or your own initialized polling object for a personal polling strategy. 104 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 105 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 106 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 107 :rtype: ~azure.core.polling.AsyncLROPoller[None] 108 :raises ~azure.core.exceptions.HttpResponseError: 109 """ 110 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 111 cls = kwargs.pop('cls', None) # type: ClsType[None] 112 lro_delay = kwargs.pop( 113 'polling_interval', 114 self._config.polling_interval 115 ) 116 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 117 if cont_token is None: 118 raw_result = await self._delete_initial( 119 resource_group_name=resource_group_name, 120 application_security_group_name=application_security_group_name, 121 cls=lambda x,y,z: x, 122 **kwargs 123 ) 124 125 kwargs.pop('error_map', None) 126 kwargs.pop('content_type', None) 127 128 def get_long_running_output(pipeline_response): 129 if cls: 130 return cls(pipeline_response, None, {}) 131 132 path_format_arguments = { 133 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 134 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 135 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 136 } 137 138 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 139 elif polling is False: polling_method = AsyncNoPolling() 140 else: polling_method = polling 141 if cont_token: 142 return AsyncLROPoller.from_continuation_token( 143 polling_method=polling_method, 144 continuation_token=cont_token, 145 client=self._client, 146 deserialization_callback=get_long_running_output 147 ) 148 else: 149 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 150 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 151 152 async def get( 153 self, 154 resource_group_name: str, 155 application_security_group_name: str, 156 **kwargs 157 ) -> "_models.ApplicationSecurityGroup": 158 """Gets information about the specified application security group. 159 160 :param resource_group_name: The name of the resource group. 161 :type resource_group_name: str 162 :param application_security_group_name: The name of the application security group. 163 :type application_security_group_name: str 164 :keyword callable cls: A custom type or function that will be passed the direct response 165 :return: ApplicationSecurityGroup, or the result of cls(response) 166 :rtype: ~azure.mgmt.network.v2018_06_01.models.ApplicationSecurityGroup 167 :raises: ~azure.core.exceptions.HttpResponseError 168 """ 169 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroup"] 170 error_map = { 171 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 172 } 173 error_map.update(kwargs.pop('error_map', {})) 174 api_version = "2018-06-01" 175 accept = "application/json" 176 177 # Construct URL 178 url = self.get.metadata['url'] # type: ignore 179 path_format_arguments = { 180 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 181 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 182 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 183 } 184 url = self._client.format_url(url, **path_format_arguments) 185 186 # Construct parameters 187 query_parameters = {} # type: Dict[str, Any] 188 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 189 190 # Construct headers 191 header_parameters = {} # type: Dict[str, Any] 192 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 193 194 request = self._client.get(url, query_parameters, header_parameters) 195 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 196 response = pipeline_response.http_response 197 198 if response.status_code not in [200]: 199 map_error(status_code=response.status_code, response=response, error_map=error_map) 200 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 201 202 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 203 204 if cls: 205 return cls(pipeline_response, deserialized, {}) 206 207 return deserialized 208 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 209 210 async def _create_or_update_initial( 211 self, 212 resource_group_name: str, 213 application_security_group_name: str, 214 parameters: "_models.ApplicationSecurityGroup", 215 **kwargs 216 ) -> "_models.ApplicationSecurityGroup": 217 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroup"] 218 error_map = { 219 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 220 } 221 error_map.update(kwargs.pop('error_map', {})) 222 api_version = "2018-06-01" 223 content_type = kwargs.pop("content_type", "application/json") 224 accept = "application/json" 225 226 # Construct URL 227 url = self._create_or_update_initial.metadata['url'] # type: ignore 228 path_format_arguments = { 229 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 230 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 231 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 232 } 233 url = self._client.format_url(url, **path_format_arguments) 234 235 # Construct parameters 236 query_parameters = {} # type: Dict[str, Any] 237 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 238 239 # Construct headers 240 header_parameters = {} # type: Dict[str, Any] 241 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 242 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 243 244 body_content_kwargs = {} # type: Dict[str, Any] 245 body_content = self._serialize.body(parameters, 'ApplicationSecurityGroup') 246 body_content_kwargs['content'] = body_content 247 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 248 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 249 response = pipeline_response.http_response 250 251 if response.status_code not in [200, 201]: 252 map_error(status_code=response.status_code, response=response, error_map=error_map) 253 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 254 255 if response.status_code == 200: 256 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 257 258 if response.status_code == 201: 259 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 260 261 if cls: 262 return cls(pipeline_response, deserialized, {}) 263 264 return deserialized 265 _create_or_update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 266 267 async def begin_create_or_update( 268 self, 269 resource_group_name: str, 270 application_security_group_name: str, 271 parameters: "_models.ApplicationSecurityGroup", 272 **kwargs 273 ) -> AsyncLROPoller["_models.ApplicationSecurityGroup"]: 274 """Creates or updates an application security group. 275 276 :param resource_group_name: The name of the resource group. 277 :type resource_group_name: str 278 :param application_security_group_name: The name of the application security group. 279 :type application_security_group_name: str 280 :param parameters: Parameters supplied to the create or update ApplicationSecurityGroup 281 operation. 282 :type parameters: ~azure.mgmt.network.v2018_06_01.models.ApplicationSecurityGroup 283 :keyword callable cls: A custom type or function that will be passed the direct response 284 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 285 :keyword polling: Pass in True if you'd like the AsyncARMPolling polling method, 286 False for no polling, or your own initialized polling object for a personal polling strategy. 287 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 288 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 289 :return: An instance of AsyncLROPoller that returns either ApplicationSecurityGroup or the result of cls(response) 290 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.network.v2018_06_01.models.ApplicationSecurityGroup] 291 :raises ~azure.core.exceptions.HttpResponseError: 292 """ 293 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 294 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroup"] 295 lro_delay = kwargs.pop( 296 'polling_interval', 297 self._config.polling_interval 298 ) 299 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 300 if cont_token is None: 301 raw_result = await self._create_or_update_initial( 302 resource_group_name=resource_group_name, 303 application_security_group_name=application_security_group_name, 304 parameters=parameters, 305 cls=lambda x,y,z: x, 306 **kwargs 307 ) 308 309 kwargs.pop('error_map', None) 310 kwargs.pop('content_type', None) 311 312 def get_long_running_output(pipeline_response): 313 deserialized = self._deserialize('ApplicationSecurityGroup', pipeline_response) 314 315 if cls: 316 return cls(pipeline_response, deserialized, {}) 317 return deserialized 318 319 path_format_arguments = { 320 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 321 'applicationSecurityGroupName': self._serialize.url("application_security_group_name", application_security_group_name, 'str'), 322 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 323 } 324 325 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 326 elif polling is False: polling_method = AsyncNoPolling() 327 else: polling_method = polling 328 if cont_token: 329 return AsyncLROPoller.from_continuation_token( 330 polling_method=polling_method, 331 continuation_token=cont_token, 332 client=self._client, 333 deserialization_callback=get_long_running_output 334 ) 335 else: 336 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 337 begin_create_or_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups/{applicationSecurityGroupName}'} # type: ignore 338 339 def list_all( 340 self, 341 **kwargs 342 ) -> AsyncIterable["_models.ApplicationSecurityGroupListResult"]: 343 """Gets all application security groups in a subscription. 344 345 :keyword callable cls: A custom type or function that will be passed the direct response 346 :return: An iterator like instance of either ApplicationSecurityGroupListResult or the result of cls(response) 347 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2018_06_01.models.ApplicationSecurityGroupListResult] 348 :raises: ~azure.core.exceptions.HttpResponseError 349 """ 350 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroupListResult"] 351 error_map = { 352 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 353 } 354 error_map.update(kwargs.pop('error_map', {})) 355 api_version = "2018-06-01" 356 accept = "application/json" 357 358 def prepare_request(next_link=None): 359 # Construct headers 360 header_parameters = {} # type: Dict[str, Any] 361 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 362 363 if not next_link: 364 # Construct URL 365 url = self.list_all.metadata['url'] # type: ignore 366 path_format_arguments = { 367 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 368 } 369 url = self._client.format_url(url, **path_format_arguments) 370 # Construct parameters 371 query_parameters = {} # type: Dict[str, Any] 372 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 373 374 request = self._client.get(url, query_parameters, header_parameters) 375 else: 376 url = next_link 377 query_parameters = {} # type: Dict[str, Any] 378 request = self._client.get(url, query_parameters, header_parameters) 379 return request 380 381 async def extract_data(pipeline_response): 382 deserialized = self._deserialize('ApplicationSecurityGroupListResult', pipeline_response) 383 list_of_elem = deserialized.value 384 if cls: 385 list_of_elem = cls(list_of_elem) 386 return deserialized.next_link or None, AsyncList(list_of_elem) 387 388 async def get_next(next_link=None): 389 request = prepare_request(next_link) 390 391 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 392 response = pipeline_response.http_response 393 394 if response.status_code not in [200]: 395 map_error(status_code=response.status_code, response=response, error_map=error_map) 396 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 397 398 return pipeline_response 399 400 return AsyncItemPaged( 401 get_next, extract_data 402 ) 403 list_all.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Network/applicationSecurityGroups'} # type: ignore 404 405 def list( 406 self, 407 resource_group_name: str, 408 **kwargs 409 ) -> AsyncIterable["_models.ApplicationSecurityGroupListResult"]: 410 """Gets all the application security groups in a resource group. 411 412 :param resource_group_name: The name of the resource group. 413 :type resource_group_name: str 414 :keyword callable cls: A custom type or function that will be passed the direct response 415 :return: An iterator like instance of either ApplicationSecurityGroupListResult or the result of cls(response) 416 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.network.v2018_06_01.models.ApplicationSecurityGroupListResult] 417 :raises: ~azure.core.exceptions.HttpResponseError 418 """ 419 cls = kwargs.pop('cls', None) # type: ClsType["_models.ApplicationSecurityGroupListResult"] 420 error_map = { 421 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 422 } 423 error_map.update(kwargs.pop('error_map', {})) 424 api_version = "2018-06-01" 425 accept = "application/json" 426 427 def prepare_request(next_link=None): 428 # Construct headers 429 header_parameters = {} # type: Dict[str, Any] 430 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 431 432 if not next_link: 433 # Construct URL 434 url = self.list.metadata['url'] # type: ignore 435 path_format_arguments = { 436 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 437 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 438 } 439 url = self._client.format_url(url, **path_format_arguments) 440 # Construct parameters 441 query_parameters = {} # type: Dict[str, Any] 442 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 443 444 request = self._client.get(url, query_parameters, header_parameters) 445 else: 446 url = next_link 447 query_parameters = {} # type: Dict[str, Any] 448 request = self._client.get(url, query_parameters, header_parameters) 449 return request 450 451 async def extract_data(pipeline_response): 452 deserialized = self._deserialize('ApplicationSecurityGroupListResult', pipeline_response) 453 list_of_elem = deserialized.value 454 if cls: 455 list_of_elem = cls(list_of_elem) 456 return deserialized.next_link or None, AsyncList(list_of_elem) 457 458 async def get_next(next_link=None): 459 request = prepare_request(next_link) 460 461 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 462 response = pipeline_response.http_response 463 464 if response.status_code not in [200]: 465 map_error(status_code=response.status_code, response=response, error_map=error_map) 466 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 467 468 return pipeline_response 469 470 return AsyncItemPaged( 471 get_next, extract_data 472 ) 473 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Network/applicationSecurityGroups'} # type: ignore 474