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 TYPE_CHECKING 9import warnings 10 11from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error 12from azure.core.paging import ItemPaged 13from azure.core.pipeline import PipelineResponse 14from azure.core.pipeline.transport import HttpRequest, HttpResponse 15from azure.core.polling import LROPoller, NoPolling, PollingMethod 16from azure.mgmt.core.exceptions import ARMErrorFormat 17from azure.mgmt.core.polling.arm_polling import ARMPolling 18 19from .. import models as _models 20 21if TYPE_CHECKING: 22 # pylint: disable=unused-import,ungrouped-imports 23 from typing import Any, Callable, Dict, Generic, Iterable, Optional, TypeVar, Union 24 25 T = TypeVar('T') 26 ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]] 27 28class LongTermRetentionManagedInstanceBackupsOperations(object): 29 """LongTermRetentionManagedInstanceBackupsOperations operations. 30 31 You should not instantiate this class directly. Instead, you should create a Client instance that 32 instantiates it for you and attaches it as an attribute. 33 34 :ivar models: Alias to model classes used in this operation group. 35 :type models: ~azure.mgmt.sql.models 36 :param client: Client for service requests. 37 :param config: Configuration of service client. 38 :param serializer: An object model serializer. 39 :param deserializer: An object model deserializer. 40 """ 41 42 models = _models 43 44 def __init__(self, client, config, serializer, deserializer): 45 self._client = client 46 self._serialize = serializer 47 self._deserialize = deserializer 48 self._config = config 49 50 def get( 51 self, 52 location_name, # type: str 53 managed_instance_name, # type: str 54 database_name, # type: str 55 backup_name, # type: str 56 **kwargs # type: Any 57 ): 58 # type: (...) -> "_models.ManagedInstanceLongTermRetentionBackup" 59 """Gets a long term retention backup for a managed database. 60 61 :param location_name: The location of the database. 62 :type location_name: str 63 :param managed_instance_name: The name of the managed instance. 64 :type managed_instance_name: str 65 :param database_name: The name of the managed database. 66 :type database_name: str 67 :param backup_name: The backup name. 68 :type backup_name: str 69 :keyword callable cls: A custom type or function that will be passed the direct response 70 :return: ManagedInstanceLongTermRetentionBackup, or the result of cls(response) 71 :rtype: ~azure.mgmt.sql.models.ManagedInstanceLongTermRetentionBackup 72 :raises: ~azure.core.exceptions.HttpResponseError 73 """ 74 cls = kwargs.pop('cls', None) # type: ClsType["_models.ManagedInstanceLongTermRetentionBackup"] 75 error_map = { 76 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 77 } 78 error_map.update(kwargs.pop('error_map', {})) 79 api_version = "2020-11-01-preview" 80 accept = "application/json" 81 82 # Construct URL 83 url = self.get.metadata['url'] # type: ignore 84 path_format_arguments = { 85 'locationName': self._serialize.url("location_name", location_name, 'str'), 86 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 87 'databaseName': self._serialize.url("database_name", database_name, 'str'), 88 'backupName': self._serialize.url("backup_name", backup_name, 'str'), 89 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, '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 = 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('ManagedInstanceLongTermRetentionBackup', pipeline_response) 110 111 if cls: 112 return cls(pipeline_response, deserialized, {}) 113 114 return deserialized 115 get.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionDatabases/{databaseName}/longTermRetentionManagedInstanceBackups/{backupName}'} # type: ignore 116 117 def _delete_initial( 118 self, 119 location_name, # type: str 120 managed_instance_name, # type: str 121 database_name, # type: str 122 backup_name, # type: str 123 **kwargs # type: Any 124 ): 125 # type: (...) -> None 126 cls = kwargs.pop('cls', None) # type: ClsType[None] 127 error_map = { 128 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 129 } 130 error_map.update(kwargs.pop('error_map', {})) 131 api_version = "2020-11-01-preview" 132 133 # Construct URL 134 url = self._delete_initial.metadata['url'] # type: ignore 135 path_format_arguments = { 136 'locationName': self._serialize.url("location_name", location_name, 'str'), 137 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 138 'databaseName': self._serialize.url("database_name", database_name, 'str'), 139 'backupName': self._serialize.url("backup_name", backup_name, 'str'), 140 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 141 } 142 url = self._client.format_url(url, **path_format_arguments) 143 144 # Construct parameters 145 query_parameters = {} # type: Dict[str, Any] 146 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 147 148 # Construct headers 149 header_parameters = {} # type: Dict[str, Any] 150 151 request = self._client.delete(url, query_parameters, header_parameters) 152 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 153 response = pipeline_response.http_response 154 155 if response.status_code not in [200, 202]: 156 map_error(status_code=response.status_code, response=response, error_map=error_map) 157 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 158 159 if cls: 160 return cls(pipeline_response, None, {}) 161 162 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionDatabases/{databaseName}/longTermRetentionManagedInstanceBackups/{backupName}'} # type: ignore 163 164 def begin_delete( 165 self, 166 location_name, # type: str 167 managed_instance_name, # type: str 168 database_name, # type: str 169 backup_name, # type: str 170 **kwargs # type: Any 171 ): 172 # type: (...) -> LROPoller[None] 173 """Deletes a long term retention backup. 174 175 :param location_name: The location of the database. 176 :type location_name: str 177 :param managed_instance_name: The name of the managed instance. 178 :type managed_instance_name: str 179 :param database_name: The name of the managed database. 180 :type database_name: str 181 :param backup_name: The backup name. 182 :type backup_name: str 183 :keyword callable cls: A custom type or function that will be passed the direct response 184 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 185 :keyword polling: By default, your polling method will be ARMPolling. 186 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 187 :paramtype polling: bool or ~azure.core.polling.PollingMethod 188 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 189 :return: An instance of LROPoller that returns either None or the result of cls(response) 190 :rtype: ~azure.core.polling.LROPoller[None] 191 :raises ~azure.core.exceptions.HttpResponseError: 192 """ 193 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 194 cls = kwargs.pop('cls', None) # type: ClsType[None] 195 lro_delay = kwargs.pop( 196 'polling_interval', 197 self._config.polling_interval 198 ) 199 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 200 if cont_token is None: 201 raw_result = self._delete_initial( 202 location_name=location_name, 203 managed_instance_name=managed_instance_name, 204 database_name=database_name, 205 backup_name=backup_name, 206 cls=lambda x,y,z: x, 207 **kwargs 208 ) 209 210 kwargs.pop('error_map', None) 211 kwargs.pop('content_type', None) 212 213 def get_long_running_output(pipeline_response): 214 if cls: 215 return cls(pipeline_response, None, {}) 216 217 path_format_arguments = { 218 'locationName': self._serialize.url("location_name", location_name, 'str'), 219 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 220 'databaseName': self._serialize.url("database_name", database_name, 'str'), 221 'backupName': self._serialize.url("backup_name", backup_name, 'str'), 222 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 223 } 224 225 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 226 elif polling is False: polling_method = NoPolling() 227 else: polling_method = polling 228 if cont_token: 229 return LROPoller.from_continuation_token( 230 polling_method=polling_method, 231 continuation_token=cont_token, 232 client=self._client, 233 deserialization_callback=get_long_running_output 234 ) 235 else: 236 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 237 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionDatabases/{databaseName}/longTermRetentionManagedInstanceBackups/{backupName}'} # type: ignore 238 239 def list_by_database( 240 self, 241 location_name, # type: str 242 managed_instance_name, # type: str 243 database_name, # type: str 244 only_latest_per_database=None, # type: Optional[bool] 245 database_state=None, # type: Optional[Union[str, "_models.DatabaseState"]] 246 **kwargs # type: Any 247 ): 248 # type: (...) -> Iterable["_models.ManagedInstanceLongTermRetentionBackupListResult"] 249 """Lists all long term retention backups for a managed database. 250 251 :param location_name: The location of the database. 252 :type location_name: str 253 :param managed_instance_name: The name of the managed instance. 254 :type managed_instance_name: str 255 :param database_name: The name of the managed database. 256 :type database_name: str 257 :param only_latest_per_database: Whether or not to only get the latest backup for each 258 database. 259 :type only_latest_per_database: bool 260 :param database_state: Whether to query against just live databases, just deleted databases, or 261 all databases. 262 :type database_state: str or ~azure.mgmt.sql.models.DatabaseState 263 :keyword callable cls: A custom type or function that will be passed the direct response 264 :return: An iterator like instance of either ManagedInstanceLongTermRetentionBackupListResult or the result of cls(response) 265 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sql.models.ManagedInstanceLongTermRetentionBackupListResult] 266 :raises: ~azure.core.exceptions.HttpResponseError 267 """ 268 cls = kwargs.pop('cls', None) # type: ClsType["_models.ManagedInstanceLongTermRetentionBackupListResult"] 269 error_map = { 270 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 271 } 272 error_map.update(kwargs.pop('error_map', {})) 273 api_version = "2020-11-01-preview" 274 accept = "application/json" 275 276 def prepare_request(next_link=None): 277 # Construct headers 278 header_parameters = {} # type: Dict[str, Any] 279 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 280 281 if not next_link: 282 # Construct URL 283 url = self.list_by_database.metadata['url'] # type: ignore 284 path_format_arguments = { 285 'locationName': self._serialize.url("location_name", location_name, 'str'), 286 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 287 'databaseName': self._serialize.url("database_name", database_name, 'str'), 288 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 289 } 290 url = self._client.format_url(url, **path_format_arguments) 291 # Construct parameters 292 query_parameters = {} # type: Dict[str, Any] 293 if only_latest_per_database is not None: 294 query_parameters['onlyLatestPerDatabase'] = self._serialize.query("only_latest_per_database", only_latest_per_database, 'bool') 295 if database_state is not None: 296 query_parameters['databaseState'] = self._serialize.query("database_state", database_state, 'str') 297 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 298 299 request = self._client.get(url, query_parameters, header_parameters) 300 else: 301 url = next_link 302 query_parameters = {} # type: Dict[str, Any] 303 request = self._client.get(url, query_parameters, header_parameters) 304 return request 305 306 def extract_data(pipeline_response): 307 deserialized = self._deserialize('ManagedInstanceLongTermRetentionBackupListResult', pipeline_response) 308 list_of_elem = deserialized.value 309 if cls: 310 list_of_elem = cls(list_of_elem) 311 return deserialized.next_link or None, iter(list_of_elem) 312 313 def get_next(next_link=None): 314 request = prepare_request(next_link) 315 316 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 317 response = pipeline_response.http_response 318 319 if response.status_code not in [200]: 320 map_error(status_code=response.status_code, response=response, error_map=error_map) 321 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 322 323 return pipeline_response 324 325 return ItemPaged( 326 get_next, extract_data 327 ) 328 list_by_database.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionDatabases/{databaseName}/longTermRetentionManagedInstanceBackups'} # type: ignore 329 330 def list_by_instance( 331 self, 332 location_name, # type: str 333 managed_instance_name, # type: str 334 only_latest_per_database=None, # type: Optional[bool] 335 database_state=None, # type: Optional[Union[str, "_models.DatabaseState"]] 336 **kwargs # type: Any 337 ): 338 # type: (...) -> Iterable["_models.ManagedInstanceLongTermRetentionBackupListResult"] 339 """Lists the long term retention backups for a given managed instance. 340 341 :param location_name: The location of the database. 342 :type location_name: str 343 :param managed_instance_name: The name of the managed instance. 344 :type managed_instance_name: str 345 :param only_latest_per_database: Whether or not to only get the latest backup for each 346 database. 347 :type only_latest_per_database: bool 348 :param database_state: Whether to query against just live databases, just deleted databases, or 349 all databases. 350 :type database_state: str or ~azure.mgmt.sql.models.DatabaseState 351 :keyword callable cls: A custom type or function that will be passed the direct response 352 :return: An iterator like instance of either ManagedInstanceLongTermRetentionBackupListResult or the result of cls(response) 353 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sql.models.ManagedInstanceLongTermRetentionBackupListResult] 354 :raises: ~azure.core.exceptions.HttpResponseError 355 """ 356 cls = kwargs.pop('cls', None) # type: ClsType["_models.ManagedInstanceLongTermRetentionBackupListResult"] 357 error_map = { 358 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 359 } 360 error_map.update(kwargs.pop('error_map', {})) 361 api_version = "2020-11-01-preview" 362 accept = "application/json" 363 364 def prepare_request(next_link=None): 365 # Construct headers 366 header_parameters = {} # type: Dict[str, Any] 367 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 368 369 if not next_link: 370 # Construct URL 371 url = self.list_by_instance.metadata['url'] # type: ignore 372 path_format_arguments = { 373 'locationName': self._serialize.url("location_name", location_name, 'str'), 374 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 375 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 376 } 377 url = self._client.format_url(url, **path_format_arguments) 378 # Construct parameters 379 query_parameters = {} # type: Dict[str, Any] 380 if only_latest_per_database is not None: 381 query_parameters['onlyLatestPerDatabase'] = self._serialize.query("only_latest_per_database", only_latest_per_database, 'bool') 382 if database_state is not None: 383 query_parameters['databaseState'] = self._serialize.query("database_state", database_state, 'str') 384 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 385 386 request = self._client.get(url, query_parameters, header_parameters) 387 else: 388 url = next_link 389 query_parameters = {} # type: Dict[str, Any] 390 request = self._client.get(url, query_parameters, header_parameters) 391 return request 392 393 def extract_data(pipeline_response): 394 deserialized = self._deserialize('ManagedInstanceLongTermRetentionBackupListResult', pipeline_response) 395 list_of_elem = deserialized.value 396 if cls: 397 list_of_elem = cls(list_of_elem) 398 return deserialized.next_link or None, iter(list_of_elem) 399 400 def get_next(next_link=None): 401 request = prepare_request(next_link) 402 403 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 404 response = pipeline_response.http_response 405 406 if response.status_code not in [200]: 407 map_error(status_code=response.status_code, response=response, error_map=error_map) 408 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 409 410 return pipeline_response 411 412 return ItemPaged( 413 get_next, extract_data 414 ) 415 list_by_instance.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionManagedInstanceBackups'} # type: ignore 416 417 def list_by_location( 418 self, 419 location_name, # type: str 420 only_latest_per_database=None, # type: Optional[bool] 421 database_state=None, # type: Optional[Union[str, "_models.DatabaseState"]] 422 **kwargs # type: Any 423 ): 424 # type: (...) -> Iterable["_models.ManagedInstanceLongTermRetentionBackupListResult"] 425 """Lists the long term retention backups for managed databases in a given location. 426 427 :param location_name: The location of the database. 428 :type location_name: str 429 :param only_latest_per_database: Whether or not to only get the latest backup for each 430 database. 431 :type only_latest_per_database: bool 432 :param database_state: Whether to query against just live databases, just deleted databases, or 433 all databases. 434 :type database_state: str or ~azure.mgmt.sql.models.DatabaseState 435 :keyword callable cls: A custom type or function that will be passed the direct response 436 :return: An iterator like instance of either ManagedInstanceLongTermRetentionBackupListResult or the result of cls(response) 437 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sql.models.ManagedInstanceLongTermRetentionBackupListResult] 438 :raises: ~azure.core.exceptions.HttpResponseError 439 """ 440 cls = kwargs.pop('cls', None) # type: ClsType["_models.ManagedInstanceLongTermRetentionBackupListResult"] 441 error_map = { 442 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 443 } 444 error_map.update(kwargs.pop('error_map', {})) 445 api_version = "2020-11-01-preview" 446 accept = "application/json" 447 448 def prepare_request(next_link=None): 449 # Construct headers 450 header_parameters = {} # type: Dict[str, Any] 451 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 452 453 if not next_link: 454 # Construct URL 455 url = self.list_by_location.metadata['url'] # type: ignore 456 path_format_arguments = { 457 'locationName': self._serialize.url("location_name", location_name, 'str'), 458 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 459 } 460 url = self._client.format_url(url, **path_format_arguments) 461 # Construct parameters 462 query_parameters = {} # type: Dict[str, Any] 463 if only_latest_per_database is not None: 464 query_parameters['onlyLatestPerDatabase'] = self._serialize.query("only_latest_per_database", only_latest_per_database, 'bool') 465 if database_state is not None: 466 query_parameters['databaseState'] = self._serialize.query("database_state", database_state, 'str') 467 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 468 469 request = self._client.get(url, query_parameters, header_parameters) 470 else: 471 url = next_link 472 query_parameters = {} # type: Dict[str, Any] 473 request = self._client.get(url, query_parameters, header_parameters) 474 return request 475 476 def extract_data(pipeline_response): 477 deserialized = self._deserialize('ManagedInstanceLongTermRetentionBackupListResult', pipeline_response) 478 list_of_elem = deserialized.value 479 if cls: 480 list_of_elem = cls(list_of_elem) 481 return deserialized.next_link or None, iter(list_of_elem) 482 483 def get_next(next_link=None): 484 request = prepare_request(next_link) 485 486 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 487 response = pipeline_response.http_response 488 489 if response.status_code not in [200]: 490 map_error(status_code=response.status_code, response=response, error_map=error_map) 491 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 492 493 return pipeline_response 494 495 return ItemPaged( 496 get_next, extract_data 497 ) 498 list_by_location.metadata = {'url': '/subscriptions/{subscriptionId}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstanceBackups'} # type: ignore 499 500 def get_by_resource_group( 501 self, 502 resource_group_name, # type: str 503 location_name, # type: str 504 managed_instance_name, # type: str 505 database_name, # type: str 506 backup_name, # type: str 507 **kwargs # type: Any 508 ): 509 # type: (...) -> "_models.ManagedInstanceLongTermRetentionBackup" 510 """Gets a long term retention backup for a managed database. 511 512 :param resource_group_name: The name of the resource group that contains the resource. You can 513 obtain this value from the Azure Resource Manager API or the portal. 514 :type resource_group_name: str 515 :param location_name: The location of the database. 516 :type location_name: str 517 :param managed_instance_name: The name of the managed instance. 518 :type managed_instance_name: str 519 :param database_name: The name of the managed database. 520 :type database_name: str 521 :param backup_name: The backup name. 522 :type backup_name: str 523 :keyword callable cls: A custom type or function that will be passed the direct response 524 :return: ManagedInstanceLongTermRetentionBackup, or the result of cls(response) 525 :rtype: ~azure.mgmt.sql.models.ManagedInstanceLongTermRetentionBackup 526 :raises: ~azure.core.exceptions.HttpResponseError 527 """ 528 cls = kwargs.pop('cls', None) # type: ClsType["_models.ManagedInstanceLongTermRetentionBackup"] 529 error_map = { 530 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 531 } 532 error_map.update(kwargs.pop('error_map', {})) 533 api_version = "2020-11-01-preview" 534 accept = "application/json" 535 536 # Construct URL 537 url = self.get_by_resource_group.metadata['url'] # type: ignore 538 path_format_arguments = { 539 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 540 'locationName': self._serialize.url("location_name", location_name, 'str'), 541 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 542 'databaseName': self._serialize.url("database_name", database_name, 'str'), 543 'backupName': self._serialize.url("backup_name", backup_name, 'str'), 544 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 545 } 546 url = self._client.format_url(url, **path_format_arguments) 547 548 # Construct parameters 549 query_parameters = {} # type: Dict[str, Any] 550 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 551 552 # Construct headers 553 header_parameters = {} # type: Dict[str, Any] 554 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 555 556 request = self._client.get(url, query_parameters, header_parameters) 557 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 558 response = pipeline_response.http_response 559 560 if response.status_code not in [200]: 561 map_error(status_code=response.status_code, response=response, error_map=error_map) 562 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 563 564 deserialized = self._deserialize('ManagedInstanceLongTermRetentionBackup', pipeline_response) 565 566 if cls: 567 return cls(pipeline_response, deserialized, {}) 568 569 return deserialized 570 get_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionDatabases/{databaseName}/longTermRetentionManagedInstanceBackups/{backupName}'} # type: ignore 571 572 def _delete_by_resource_group_initial( 573 self, 574 resource_group_name, # type: str 575 location_name, # type: str 576 managed_instance_name, # type: str 577 database_name, # type: str 578 backup_name, # type: str 579 **kwargs # type: Any 580 ): 581 # type: (...) -> None 582 cls = kwargs.pop('cls', None) # type: ClsType[None] 583 error_map = { 584 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 585 } 586 error_map.update(kwargs.pop('error_map', {})) 587 api_version = "2020-11-01-preview" 588 589 # Construct URL 590 url = self._delete_by_resource_group_initial.metadata['url'] # type: ignore 591 path_format_arguments = { 592 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 593 'locationName': self._serialize.url("location_name", location_name, 'str'), 594 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 595 'databaseName': self._serialize.url("database_name", database_name, 'str'), 596 'backupName': self._serialize.url("backup_name", backup_name, 'str'), 597 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 598 } 599 url = self._client.format_url(url, **path_format_arguments) 600 601 # Construct parameters 602 query_parameters = {} # type: Dict[str, Any] 603 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 604 605 # Construct headers 606 header_parameters = {} # type: Dict[str, Any] 607 608 request = self._client.delete(url, query_parameters, header_parameters) 609 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 610 response = pipeline_response.http_response 611 612 if response.status_code not in [200, 202]: 613 map_error(status_code=response.status_code, response=response, error_map=error_map) 614 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 615 616 if cls: 617 return cls(pipeline_response, None, {}) 618 619 _delete_by_resource_group_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionDatabases/{databaseName}/longTermRetentionManagedInstanceBackups/{backupName}'} # type: ignore 620 621 def begin_delete_by_resource_group( 622 self, 623 resource_group_name, # type: str 624 location_name, # type: str 625 managed_instance_name, # type: str 626 database_name, # type: str 627 backup_name, # type: str 628 **kwargs # type: Any 629 ): 630 # type: (...) -> LROPoller[None] 631 """Deletes a long term retention backup. 632 633 :param resource_group_name: The name of the resource group that contains the resource. You can 634 obtain this value from the Azure Resource Manager API or the portal. 635 :type resource_group_name: str 636 :param location_name: The location of the database. 637 :type location_name: str 638 :param managed_instance_name: The name of the managed instance. 639 :type managed_instance_name: str 640 :param database_name: The name of the managed database. 641 :type database_name: str 642 :param backup_name: The backup name. 643 :type backup_name: str 644 :keyword callable cls: A custom type or function that will be passed the direct response 645 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 646 :keyword polling: By default, your polling method will be ARMPolling. 647 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 648 :paramtype polling: bool or ~azure.core.polling.PollingMethod 649 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 650 :return: An instance of LROPoller that returns either None or the result of cls(response) 651 :rtype: ~azure.core.polling.LROPoller[None] 652 :raises ~azure.core.exceptions.HttpResponseError: 653 """ 654 polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] 655 cls = kwargs.pop('cls', None) # type: ClsType[None] 656 lro_delay = kwargs.pop( 657 'polling_interval', 658 self._config.polling_interval 659 ) 660 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 661 if cont_token is None: 662 raw_result = self._delete_by_resource_group_initial( 663 resource_group_name=resource_group_name, 664 location_name=location_name, 665 managed_instance_name=managed_instance_name, 666 database_name=database_name, 667 backup_name=backup_name, 668 cls=lambda x,y,z: x, 669 **kwargs 670 ) 671 672 kwargs.pop('error_map', None) 673 kwargs.pop('content_type', None) 674 675 def get_long_running_output(pipeline_response): 676 if cls: 677 return cls(pipeline_response, None, {}) 678 679 path_format_arguments = { 680 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 681 'locationName': self._serialize.url("location_name", location_name, 'str'), 682 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 683 'databaseName': self._serialize.url("database_name", database_name, 'str'), 684 'backupName': self._serialize.url("backup_name", backup_name, 'str'), 685 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 686 } 687 688 if polling is True: polling_method = ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 689 elif polling is False: polling_method = NoPolling() 690 else: polling_method = polling 691 if cont_token: 692 return LROPoller.from_continuation_token( 693 polling_method=polling_method, 694 continuation_token=cont_token, 695 client=self._client, 696 deserialization_callback=get_long_running_output 697 ) 698 else: 699 return LROPoller(self._client, raw_result, get_long_running_output, polling_method) 700 begin_delete_by_resource_group.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionDatabases/{databaseName}/longTermRetentionManagedInstanceBackups/{backupName}'} # type: ignore 701 702 def list_by_resource_group_database( 703 self, 704 resource_group_name, # type: str 705 location_name, # type: str 706 managed_instance_name, # type: str 707 database_name, # type: str 708 only_latest_per_database=None, # type: Optional[bool] 709 database_state=None, # type: Optional[Union[str, "_models.DatabaseState"]] 710 **kwargs # type: Any 711 ): 712 # type: (...) -> Iterable["_models.ManagedInstanceLongTermRetentionBackupListResult"] 713 """Lists all long term retention backups for a managed database. 714 715 :param resource_group_name: The name of the resource group that contains the resource. You can 716 obtain this value from the Azure Resource Manager API or the portal. 717 :type resource_group_name: str 718 :param location_name: The location of the database. 719 :type location_name: str 720 :param managed_instance_name: The name of the managed instance. 721 :type managed_instance_name: str 722 :param database_name: The name of the managed database. 723 :type database_name: str 724 :param only_latest_per_database: Whether or not to only get the latest backup for each 725 database. 726 :type only_latest_per_database: bool 727 :param database_state: Whether to query against just live databases, just deleted databases, or 728 all databases. 729 :type database_state: str or ~azure.mgmt.sql.models.DatabaseState 730 :keyword callable cls: A custom type or function that will be passed the direct response 731 :return: An iterator like instance of either ManagedInstanceLongTermRetentionBackupListResult or the result of cls(response) 732 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sql.models.ManagedInstanceLongTermRetentionBackupListResult] 733 :raises: ~azure.core.exceptions.HttpResponseError 734 """ 735 cls = kwargs.pop('cls', None) # type: ClsType["_models.ManagedInstanceLongTermRetentionBackupListResult"] 736 error_map = { 737 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 738 } 739 error_map.update(kwargs.pop('error_map', {})) 740 api_version = "2020-11-01-preview" 741 accept = "application/json" 742 743 def prepare_request(next_link=None): 744 # Construct headers 745 header_parameters = {} # type: Dict[str, Any] 746 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 747 748 if not next_link: 749 # Construct URL 750 url = self.list_by_resource_group_database.metadata['url'] # type: ignore 751 path_format_arguments = { 752 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 753 'locationName': self._serialize.url("location_name", location_name, 'str'), 754 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 755 'databaseName': self._serialize.url("database_name", database_name, 'str'), 756 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 757 } 758 url = self._client.format_url(url, **path_format_arguments) 759 # Construct parameters 760 query_parameters = {} # type: Dict[str, Any] 761 if only_latest_per_database is not None: 762 query_parameters['onlyLatestPerDatabase'] = self._serialize.query("only_latest_per_database", only_latest_per_database, 'bool') 763 if database_state is not None: 764 query_parameters['databaseState'] = self._serialize.query("database_state", database_state, 'str') 765 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 766 767 request = self._client.get(url, query_parameters, header_parameters) 768 else: 769 url = next_link 770 query_parameters = {} # type: Dict[str, Any] 771 request = self._client.get(url, query_parameters, header_parameters) 772 return request 773 774 def extract_data(pipeline_response): 775 deserialized = self._deserialize('ManagedInstanceLongTermRetentionBackupListResult', pipeline_response) 776 list_of_elem = deserialized.value 777 if cls: 778 list_of_elem = cls(list_of_elem) 779 return deserialized.next_link or None, iter(list_of_elem) 780 781 def get_next(next_link=None): 782 request = prepare_request(next_link) 783 784 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 785 response = pipeline_response.http_response 786 787 if response.status_code not in [200]: 788 map_error(status_code=response.status_code, response=response, error_map=error_map) 789 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 790 791 return pipeline_response 792 793 return ItemPaged( 794 get_next, extract_data 795 ) 796 list_by_resource_group_database.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionDatabases/{databaseName}/longTermRetentionManagedInstanceBackups'} # type: ignore 797 798 def list_by_resource_group_instance( 799 self, 800 resource_group_name, # type: str 801 location_name, # type: str 802 managed_instance_name, # type: str 803 only_latest_per_database=None, # type: Optional[bool] 804 database_state=None, # type: Optional[Union[str, "_models.DatabaseState"]] 805 **kwargs # type: Any 806 ): 807 # type: (...) -> Iterable["_models.ManagedInstanceLongTermRetentionBackupListResult"] 808 """Lists the long term retention backups for a given managed instance. 809 810 :param resource_group_name: The name of the resource group that contains the resource. You can 811 obtain this value from the Azure Resource Manager API or the portal. 812 :type resource_group_name: str 813 :param location_name: The location of the database. 814 :type location_name: str 815 :param managed_instance_name: The name of the managed instance. 816 :type managed_instance_name: str 817 :param only_latest_per_database: Whether or not to only get the latest backup for each 818 database. 819 :type only_latest_per_database: bool 820 :param database_state: Whether to query against just live databases, just deleted databases, or 821 all databases. 822 :type database_state: str or ~azure.mgmt.sql.models.DatabaseState 823 :keyword callable cls: A custom type or function that will be passed the direct response 824 :return: An iterator like instance of either ManagedInstanceLongTermRetentionBackupListResult or the result of cls(response) 825 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sql.models.ManagedInstanceLongTermRetentionBackupListResult] 826 :raises: ~azure.core.exceptions.HttpResponseError 827 """ 828 cls = kwargs.pop('cls', None) # type: ClsType["_models.ManagedInstanceLongTermRetentionBackupListResult"] 829 error_map = { 830 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 831 } 832 error_map.update(kwargs.pop('error_map', {})) 833 api_version = "2020-11-01-preview" 834 accept = "application/json" 835 836 def prepare_request(next_link=None): 837 # Construct headers 838 header_parameters = {} # type: Dict[str, Any] 839 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 840 841 if not next_link: 842 # Construct URL 843 url = self.list_by_resource_group_instance.metadata['url'] # type: ignore 844 path_format_arguments = { 845 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 846 'locationName': self._serialize.url("location_name", location_name, 'str'), 847 'managedInstanceName': self._serialize.url("managed_instance_name", managed_instance_name, 'str'), 848 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 849 } 850 url = self._client.format_url(url, **path_format_arguments) 851 # Construct parameters 852 query_parameters = {} # type: Dict[str, Any] 853 if only_latest_per_database is not None: 854 query_parameters['onlyLatestPerDatabase'] = self._serialize.query("only_latest_per_database", only_latest_per_database, 'bool') 855 if database_state is not None: 856 query_parameters['databaseState'] = self._serialize.query("database_state", database_state, 'str') 857 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 858 859 request = self._client.get(url, query_parameters, header_parameters) 860 else: 861 url = next_link 862 query_parameters = {} # type: Dict[str, Any] 863 request = self._client.get(url, query_parameters, header_parameters) 864 return request 865 866 def extract_data(pipeline_response): 867 deserialized = self._deserialize('ManagedInstanceLongTermRetentionBackupListResult', pipeline_response) 868 list_of_elem = deserialized.value 869 if cls: 870 list_of_elem = cls(list_of_elem) 871 return deserialized.next_link or None, iter(list_of_elem) 872 873 def get_next(next_link=None): 874 request = prepare_request(next_link) 875 876 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 877 response = pipeline_response.http_response 878 879 if response.status_code not in [200]: 880 map_error(status_code=response.status_code, response=response, error_map=error_map) 881 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 882 883 return pipeline_response 884 885 return ItemPaged( 886 get_next, extract_data 887 ) 888 list_by_resource_group_instance.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstances/{managedInstanceName}/longTermRetentionManagedInstanceBackups'} # type: ignore 889 890 def list_by_resource_group_location( 891 self, 892 resource_group_name, # type: str 893 location_name, # type: str 894 only_latest_per_database=None, # type: Optional[bool] 895 database_state=None, # type: Optional[Union[str, "_models.DatabaseState"]] 896 **kwargs # type: Any 897 ): 898 # type: (...) -> Iterable["_models.ManagedInstanceLongTermRetentionBackupListResult"] 899 """Lists the long term retention backups for managed databases in a given location. 900 901 :param resource_group_name: The name of the resource group that contains the resource. You can 902 obtain this value from the Azure Resource Manager API or the portal. 903 :type resource_group_name: str 904 :param location_name: The location of the database. 905 :type location_name: str 906 :param only_latest_per_database: Whether or not to only get the latest backup for each 907 database. 908 :type only_latest_per_database: bool 909 :param database_state: Whether to query against just live databases, just deleted databases, or 910 all databases. 911 :type database_state: str or ~azure.mgmt.sql.models.DatabaseState 912 :keyword callable cls: A custom type or function that will be passed the direct response 913 :return: An iterator like instance of either ManagedInstanceLongTermRetentionBackupListResult or the result of cls(response) 914 :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.sql.models.ManagedInstanceLongTermRetentionBackupListResult] 915 :raises: ~azure.core.exceptions.HttpResponseError 916 """ 917 cls = kwargs.pop('cls', None) # type: ClsType["_models.ManagedInstanceLongTermRetentionBackupListResult"] 918 error_map = { 919 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 920 } 921 error_map.update(kwargs.pop('error_map', {})) 922 api_version = "2020-11-01-preview" 923 accept = "application/json" 924 925 def prepare_request(next_link=None): 926 # Construct headers 927 header_parameters = {} # type: Dict[str, Any] 928 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 929 930 if not next_link: 931 # Construct URL 932 url = self.list_by_resource_group_location.metadata['url'] # type: ignore 933 path_format_arguments = { 934 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str'), 935 'locationName': self._serialize.url("location_name", location_name, 'str'), 936 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 937 } 938 url = self._client.format_url(url, **path_format_arguments) 939 # Construct parameters 940 query_parameters = {} # type: Dict[str, Any] 941 if only_latest_per_database is not None: 942 query_parameters['onlyLatestPerDatabase'] = self._serialize.query("only_latest_per_database", only_latest_per_database, 'bool') 943 if database_state is not None: 944 query_parameters['databaseState'] = self._serialize.query("database_state", database_state, 'str') 945 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 946 947 request = self._client.get(url, query_parameters, header_parameters) 948 else: 949 url = next_link 950 query_parameters = {} # type: Dict[str, Any] 951 request = self._client.get(url, query_parameters, header_parameters) 952 return request 953 954 def extract_data(pipeline_response): 955 deserialized = self._deserialize('ManagedInstanceLongTermRetentionBackupListResult', pipeline_response) 956 list_of_elem = deserialized.value 957 if cls: 958 list_of_elem = cls(list_of_elem) 959 return deserialized.next_link or None, iter(list_of_elem) 960 961 def get_next(next_link=None): 962 request = prepare_request(next_link) 963 964 pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) 965 response = pipeline_response.http_response 966 967 if response.status_code not in [200]: 968 map_error(status_code=response.status_code, response=response, error_map=error_map) 969 raise HttpResponseError(response=response, error_format=ARMErrorFormat) 970 971 return pipeline_response 972 973 return ItemPaged( 974 get_next, extract_data 975 ) 976 list_by_resource_group_location.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Sql/locations/{locationName}/longTermRetentionManagedInstanceBackups'} # type: ignore 977