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