# coding=utf-8 # -------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for license information. # Code generated by Microsoft (R) Python Code Generator. # Changes may cause incorrect behavior and will be lost if the code is regenerated. # -------------------------------------------------------------------------- from copy import deepcopy import sys from typing import Any, Optional, TYPE_CHECKING, cast from azure.core.pipeline import policies from azure.core.rest import HttpRequest, HttpResponse from azure.core.settings import settings from azure.mgmt.core import ARMPipelineClient from azure.mgmt.core.policies import ARMAutoResourceProviderRegistrationPolicy from azure.mgmt.core.tools import get_arm_endpoints from ._configuration import RecoveryServicesClientConfiguration from ._utils.serialization import Deserializer, Serializer from .operations import ( DeletedVaultsOperations, Operations, PrivateLinkResourcesOperations, RecoveryServicesOperations, RegisteredIdentitiesOperations, ReplicationUsagesOperations, UsagesOperations, VaultCertificatesOperations, VaultExtendedInfoOperations, VaultsOperations, _RecoveryServicesClientOperationsMixin, ) if sys.version_info >= (3, 11): from typing import Self else: from typing_extensions import Self # type: ignore if TYPE_CHECKING: from azure.core import AzureClouds from azure.core.credentials import TokenCredential class RecoveryServicesClient(_RecoveryServicesClientOperationsMixin): # pylint: disable=too-many-instance-attributes """RecoveryServicesClient. :ivar operations: Operations operations :vartype operations: azure.mgmt.recoveryservices.operations.Operations :ivar vaults: VaultsOperations operations :vartype vaults: azure.mgmt.recoveryservices.operations.VaultsOperations :ivar deleted_vaults: DeletedVaultsOperations operations :vartype deleted_vaults: azure.mgmt.recoveryservices.operations.DeletedVaultsOperations :ivar private_link_resources: PrivateLinkResourcesOperations operations :vartype private_link_resources: azure.mgmt.recoveryservices.operations.PrivateLinkResourcesOperations :ivar vault_certificates: VaultCertificatesOperations operations :vartype vault_certificates: azure.mgmt.recoveryservices.operations.VaultCertificatesOperations :ivar registered_identities: RegisteredIdentitiesOperations operations :vartype registered_identities: azure.mgmt.recoveryservices.operations.RegisteredIdentitiesOperations :ivar replication_usages: ReplicationUsagesOperations operations :vartype replication_usages: azure.mgmt.recoveryservices.operations.ReplicationUsagesOperations :ivar usages: UsagesOperations operations :vartype usages: azure.mgmt.recoveryservices.operations.UsagesOperations :ivar vault_extended_info: VaultExtendedInfoOperations operations :vartype vault_extended_info: azure.mgmt.recoveryservices.operations.VaultExtendedInfoOperations :ivar recovery_services: RecoveryServicesOperations operations :vartype recovery_services: azure.mgmt.recoveryservices.operations.RecoveryServicesOperations :param credential: Credential used to authenticate requests to the service. Required. :type credential: ~azure.core.credentials.TokenCredential :param subscription_id: The ID of the target subscription. The value must be an UUID. Required. :type subscription_id: str :param base_url: Service host. Default value is None. :type base_url: str :keyword cloud_setting: The cloud setting for which to get the ARM endpoint. Default value is None. :paramtype cloud_setting: ~azure.core.AzureClouds :keyword api_version: The API version to use for this operation. Known values are "2025-08-01" and None. Default value is None. If not set, the operation's default API version will be used. Note that overriding this default value may result in unsupported behavior. :paramtype api_version: str :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. """ def __init__( self, credential: "TokenCredential", subscription_id: str, base_url: Optional[str] = None, *, cloud_setting: Optional["AzureClouds"] = None, **kwargs: Any ) -> None: _endpoint = "{endpoint}" _cloud = cloud_setting or settings.current.azure_cloud # type: ignore _endpoints = get_arm_endpoints(_cloud) if not base_url: base_url = _endpoints["resource_manager"] credential_scopes = kwargs.pop("credential_scopes", _endpoints["credential_scopes"]) self._config = RecoveryServicesClientConfiguration( credential=credential, subscription_id=subscription_id, base_url=cast(str, base_url), cloud_setting=cloud_setting, credential_scopes=credential_scopes, **kwargs ) _policies = kwargs.pop("policies", None) if _policies is None: _policies = [ policies.RequestIdPolicy(**kwargs), self._config.headers_policy, self._config.user_agent_policy, self._config.proxy_policy, policies.ContentDecodePolicy(**kwargs), ARMAutoResourceProviderRegistrationPolicy(), self._config.redirect_policy, self._config.retry_policy, self._config.authentication_policy, self._config.custom_hook_policy, self._config.logging_policy, policies.DistributedTracingPolicy(**kwargs), policies.SensitiveHeaderCleanupPolicy(**kwargs) if self._config.redirect_policy else None, self._config.http_logging_policy, ] self._client: ARMPipelineClient = ARMPipelineClient(base_url=cast(str, _endpoint), policies=_policies, **kwargs) self._serialize = Serializer() self._deserialize = Deserializer() self._serialize.client_side_validation = False self.operations = Operations(self._client, self._config, self._serialize, self._deserialize) self.vaults = VaultsOperations(self._client, self._config, self._serialize, self._deserialize) self.deleted_vaults = DeletedVaultsOperations(self._client, self._config, self._serialize, self._deserialize) self.private_link_resources = PrivateLinkResourcesOperations( self._client, self._config, self._serialize, self._deserialize ) self.vault_certificates = VaultCertificatesOperations( self._client, self._config, self._serialize, self._deserialize ) self.registered_identities = RegisteredIdentitiesOperations( self._client, self._config, self._serialize, self._deserialize ) self.replication_usages = ReplicationUsagesOperations( self._client, self._config, self._serialize, self._deserialize ) self.usages = UsagesOperations(self._client, self._config, self._serialize, self._deserialize) self.vault_extended_info = VaultExtendedInfoOperations( self._client, self._config, self._serialize, self._deserialize ) self.recovery_services = RecoveryServicesOperations( self._client, self._config, self._serialize, self._deserialize ) def send_request(self, request: HttpRequest, *, stream: bool = False, **kwargs: Any) -> HttpResponse: """Runs the network request through the client's chained policies. >>> from azure.core.rest import HttpRequest >>> request = HttpRequest("GET", "https://www.example.org/") >>> response = client.send_request(request) For more information on this code flow, see https://aka.ms/azsdk/dpcodegen/python/send_request :param request: The network request you want to make. Required. :type request: ~azure.core.rest.HttpRequest :keyword bool stream: Whether the response payload will be streamed. Defaults to False. :return: The response of your network call. Does not do error handling on your response. :rtype: ~azure.core.rest.HttpResponse """ request_copy = deepcopy(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } request_copy.url = self._client.format_url(request_copy.url, **path_format_arguments) return self._client.send_request(request_copy, stream=stream, **kwargs) # type: ignore def close(self) -> None: self._client.close() def __enter__(self) -> Self: self._client.__enter__() return self def __exit__(self, *exc_details: Any) -> None: self._client.__exit__(*exc_details)