Azure Recovery Services

from azure.mgmt.recoveryservices import RecoveryServicesClient
from azure.identity import DefaultAzureCredential

from typing import Optional, Union, List, Dict, IO
from azure.core.polling import LROPoller
from azure.core.paging import ItemPaged
from azure.mgmt.recoveryservices.models import (
    Vault, VaultProperties, Sku, SkuName, IdentityData,
    ResourceIdentityType, PatchVault, VaultUsage, ReplicationUsage,
    CertificateRequest, VaultCertificateResponse, PrivateLinkResource,
    VaultExtendedInfoResource, CheckNameAvailabilityParameters,
    CheckNameAvailabilityResult, ResourceCapabilities, CapabilitiesResponse
)

import os
from azure.identity import DefaultAzureCredential
from azure.mgmt.recoveryservices import RecoveryServicesClient

# Initialize client with default credential
credential = DefaultAzureCredential()
subscription_id = os.environ["AZURE_SUBSCRIPTION_ID"]
client = RecoveryServicesClient(credential, subscription_id)

# List all Recovery Services vaults in subscription
vaults = client.vaults.list_by_subscription_id()
for vault in vaults:
    print(f"Vault: {vault.name}, Location: {vault.location}")

# Get a specific vault
resource_group = "my-resource-group"
vault_name = "my-recovery-vault"
vault = client.vaults.get(resource_group, vault_name)
print(f"Vault SKU: {vault.sku.name}")

# Create a new vault
from azure.mgmt.recoveryservices.models import Vault, Sku, SkuName

new_vault = Vault(
    location="eastus",
    sku=Sku(name=SkuName.STANDARD)
)

create_operation = client.vaults.begin_create_or_update(
    resource_group, "new-vault", new_vault
)
created_vault = create_operation.result()
print(f"Created vault: {created_vault.name}")

def list_by_subscription_id(**kwargs) -> ItemPaged[Vault]: ...
def list_by_resource_group(resource_group_name: str, **kwargs) -> ItemPaged[Vault]: ...
def get(resource_group_name: str, vault_name: str, **kwargs) -> Vault: ...
def begin_create_or_update(resource_group_name: str, vault_name: str, vault: Union[Vault, IO[bytes]], x_ms_authorization_auxiliary: Optional[str] = None, **kwargs) -> LROPoller[Vault]: ...
def begin_delete(resource_group_name: str, vault_name: str, **kwargs) -> LROPoller[None]: ...
def begin_update(resource_group_name: str, vault_name: str, vault: Union[PatchVault, IO[bytes]], x_ms_authorization_auxiliary: Optional[str] = None, **kwargs) -> LROPoller[Vault]: ...

def create(resource_group_name: str, vault_name: str, certificate_name: str, certificate_request: Union[CertificateRequest, IO[bytes]], **kwargs) -> VaultCertificateResponse: ...

def list(resource_group_name: str, vault_name: str, **kwargs) -> ItemPaged[PrivateLinkResource]: ...
def get(resource_group_name: str, vault_name: str, private_link_resource_name: str, **kwargs) -> PrivateLinkResource: ...

def list_by_vaults(resource_group_name: str, vault_name: str, **kwargs) -> ItemPaged[VaultUsage]: ...
def list(resource_group_name: str, vault_name: str, **kwargs) -> ItemPaged[ReplicationUsage]: ...

def get(resource_group_name: str, vault_name: str, **kwargs) -> VaultExtendedInfoResource: ...
def create_or_update(resource_group_name: str, vault_name: str, resource_extended_info_details: Union[VaultExtendedInfoResource, IO[bytes]], **kwargs) -> VaultExtendedInfoResource: ...
def update(resource_group_name: str, vault_name: str, resource_extended_info_details: Union[VaultExtendedInfoResource, IO[bytes]], **kwargs) -> VaultExtendedInfoResource: ...

def check_name_availability(resource_group_name: str, location: str, input: Union[CheckNameAvailabilityParameters, IO[bytes]], **kwargs) -> CheckNameAvailabilityResult: ...
def capabilities(location: str, input: Union[ResourceCapabilities, IO[bytes]], **kwargs) -> CapabilitiesResponse: ...

def delete(resource_group_name: str, vault_name: str, identity_name: str, **kwargs) -> None: ...

def list(**kwargs) -> ItemPaged[ClientDiscoveryValueForSingleApi]: ...

class RecoveryServicesClient:
    """
    Recovery Services Client.
    
    Parameters:
    - credential: TokenCredential - Credential needed for the client to connect to Azure
    - subscription_id: str - The ID of the target subscription
    - base_url: Optional[str] - Service URL
    - api_version: str - Api Version (default: "2025-02-01")
    - polling_interval: int - Default waiting time between two polls for LRO operations
    """
    
    def __init__(self, credential: TokenCredential, subscription_id: str, base_url: Optional[str] = None, **kwargs): ...
    def close(self) -> None: ...
    def __enter__(self) -> Self: ...
    def __exit__(self, *exc_details) -> None: ...

class Vault(TrackedResource):
    """
    Resource information, as returned by the resource provider.
    
    Parameters:
    - location: str - Resource location
    - tags: Optional[Dict[str, str]] - Resource tags
    - identity: Optional[IdentityData] - Managed service identity
    - properties: Optional[VaultProperties] - Properties of the vault
    - sku: Optional[Sku] - Identifies the unique system identifier for each Azure resource
    - system_data: Optional[SystemData] - Metadata pertaining to creation and last modification
    """

class Sku:
    """
    Identifies the unique system identifier for each Azure resource.
    
    Parameters:
    - name: Union[str, SkuName] - The Sku name (RS0, Standard)
    - tier: Optional[str] - The Sku tier
    - family: Optional[str] - The sku family
    - size: Optional[str] - The sku size
    - capacity: Optional[str] - The sku capacity
    """

class SkuName(str, Enum):
    RS0 = "RS0"
    STANDARD = "Standard"

class IdentityData:
    """
    Identity for the resource.
    
    Parameters:
    - principal_id: Optional[str] - The principal ID of resource identity
    - tenant_id: Optional[str] - The tenant ID of resource
    - type: Union[str, ResourceIdentityType] - The identity type
    - user_assigned_identities: Optional[Dict[str, UserIdentity]] - UserAssignedIdentities
    """

class ResourceIdentityType(str, Enum):
    SYSTEM_ASSIGNED = "SystemAssigned"
    USER_ASSIGNED = "UserAssigned" 
    SYSTEM_ASSIGNED_USER_ASSIGNED = "SystemAssigned, UserAssigned"
    NONE = "None"

class VaultUsage:
    """
    Vault usage information.
    
    Parameters:
    - unit: Optional[Union[str, UsagesUnit]] - Unit of the usage statistic
    - quota_period: Optional[str] - Quota period of the usage statistic
    - next_reset_time: Optional[datetime] - Next reset time of the usage statistic
    - current_value: Optional[int] - Current value of the usage statistic
    - limit: Optional[int] - Limit of the usage statistic
    - name: Optional[NameInfo] - Name of the usage statistic
    """

class ReplicationUsage:
    """
    Replication usage information.
    
    Parameters:
    - monitored_item_count: Optional[int] - Number of replication protected items
    - protected_item_count: Optional[int] - Number of replication protected VMs
    - recovery_plan_count: Optional[int] - Number of replication recovery plans
    - registered_servers_count: Optional[int] - Number of servers registered to vault
    - job_failure_count_in_day: Optional[int] - Number of replication jobs failed in a day
    """

class UsagesUnit(str, Enum):
    COUNT = "Count"
    BYTES = "Bytes"
    SECONDS = "Seconds"
    PERCENT = "Percent"
    COUNT_PER_SECOND = "CountPerSecond"
    BYTES_PER_SECOND = "BytesPerSecond"

class NameInfo:
    """
    The name of the usage.
    
    Parameters:
    - value: Optional[str] - Value of the usage
    - localized_value: Optional[str] - Localized value of the usage
    """

### Certificate Models

```python { .api }
class CertificateRequest:
    """
    Certificate request for vault.
    
    Parameters:
    - properties: Optional[RawCertificateData] - Raw certificate data
    """

class VaultCertificateResponse:
    """
    Certificate response for vault.
    
    Parameters:
    - name: Optional[str] - Certificate name
    - type: Optional[str] - Certificate type
    - id: Optional[str] - Certificate resource Id
    - properties: Optional[ResourceCertificateDetails] - Certificate details
    """

class RawCertificateData:
    """
    Raw certificate data.
    
    Parameters:
    - certificate: Optional[bytes] - The base64 encoded certificate raw data string
    - auth_type: Optional[Union[str, AuthType]] - Authentication type
    """

class PrivateLinkResource:
    """
    Information of the private link resource.
    
    Parameters:
    - group_id: Optional[str] - e.g. f9ad6492-33d4-4690-9999-6bfd52a0d081 (Backup) or f9ad6492-33d4-4690-9999-6bfd52a0d082 (SiteRecovery)
    - required_members: Optional[List[str]] - [BackupFabric, BackupProtection, BackupSecurityPin, VaultSettings]
    - required_zone_names: Optional[List[str]] - The private link resource Private link DNS zone name
    """

class VaultExtendedInfoResource:
    """
    Vault extended information.
    
    Parameters:
    - integrity_key: Optional[str] - Integrity key
    - encryption_key: Optional[str] - Encryption key
    - encryption_key_thumbprint: Optional[str] - Encryption key thumbprint
    - algorithm: Optional[str] - Algorithm for vault extended info
    """

class CheckNameAvailabilityParameters:
    """
    Resource name availability input parameters.
    
    Parameters:
    - type: Optional[str] - The resource type
    - name: Optional[str] - The resource name
    """

class CheckNameAvailabilityResult:
    """
    Response for check name availability API.
    
    Parameters:
    - name_available: Optional[bool] - Gets a boolean value that indicates availability
    - reason: Optional[str] - Gets the reason that a name is not available
    - message: Optional[str] - Gets an error message explaining the reason value
    """

class ResourceCapabilities:
    """
    Input to get capabilities API.
    
    Parameters:
    - type: str - Describes the Resource type
    - properties: Optional[ResourceCapabilitiesBase] - Describes the Resource properties
    """

class CapabilitiesResponse:
    """
    Response for get capabilities API.
    
    Parameters:
    - type: Optional[str] - Describes the Resource type
    - properties: Optional[CapabilitiesResponseProperties] - Describes the Resource properties
    """

class ClientDiscoveryValueForSingleApi:
    """
    Available operation details.
    
    Parameters:
    - name: Optional[str] - Name of the operation
    - display: Optional[ClientDiscoveryDisplay] - Contains the localized display information
    - origin: Optional[str] - The intended executor of the operation
    - properties: Optional[ClientDiscoveryForProperties] - Properties of the operation
    """

from azure.core.exceptions import (
    ClientAuthenticationError,
    HttpResponseError,
    ResourceExistsError,
    ResourceNotFoundError,
    ResourceNotModifiedError
)

try:
    vault = client.vaults.get("resource-group", "vault-name")
except ResourceNotFoundError:
    print("Vault not found")
except ClientAuthenticationError:
    print("Authentication failed")
except HttpResponseError as e:
    print(f"HTTP error: {e.status_code} - {e.message}")