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