# pylint: disable=too-many-lines,too-many-statements
# 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) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
from io import IOBase
from typing import Any, Callable, Dict, IO, Optional, TypeVar, Union, overload
from azure.core.exceptions import (
ClientAuthenticationError,
HttpResponseError,
ResourceExistsError,
ResourceNotFoundError,
ResourceNotModifiedError,
map_error,
)
from azure.core.pipeline import PipelineResponse
from azure.core.pipeline.transport import HttpResponse
from azure.core.rest import HttpRequest
from azure.core.tracing.decorator import distributed_trace
from azure.core.utils import case_insensitive_dict
from azure.mgmt.core.exceptions import ARMErrorFormat
from .. import models as _models
from .._serialization import Serializer
from .._vendor import RecoveryServicesClientMixinABC, _convert_request
T = TypeVar("T")
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
_SERIALIZER = Serializer()
_SERIALIZER.client_side_validation = False
def build_check_name_availability_request(
resource_group_name: str, location: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-04-01"))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.RecoveryServices/locations/{location}/checkNameAvailability",
) # pylint: disable=line-too-long
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str", min_length=1),
"resourceGroupName": _SERIALIZER.url(
"resource_group_name", resource_group_name, "str", max_length=90, min_length=1
),
"location": _SERIALIZER.url("location", location, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
if content_type is not None:
_headers["Content-Type"] = _SERIALIZER.header("content_type", content_type, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, **kwargs)
def build_capabilities_request(location: str, subscription_id: str, **kwargs: Any) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-04-01"))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = kwargs.pop(
"template_url",
"/subscriptions/{subscriptionId}/providers/Microsoft.RecoveryServices/locations/{location}/capabilities",
) # pylint: disable=line-too-long
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str", min_length=1),
"location": _SERIALIZER.url("location", location, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
if content_type is not None:
_headers["Content-Type"] = _SERIALIZER.header("content_type", content_type, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, **kwargs)
[docs]class RecoveryServicesOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.recoveryservices.RecoveryServicesClient`'s
:attr:`recovery_services` attribute.
"""
models = _models
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client = input_args.pop(0) if input_args else kwargs.pop("client")
self._config = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer")
@overload
def check_name_availability(
self,
resource_group_name: str,
location: str,
input: _models.CheckNameAvailabilityParameters,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.CheckNameAvailabilityResult:
"""API to check for resource name availability.
A name is available if no other resource exists that has the same SubscriptionId, Resource Name
and Type
or if one or more such resources exist, each of these must be GC'd and their time of deletion
be more than 24 Hours Ago.
API to check for resource name availability.
A name is available if no other resource exists that has the same SubscriptionId, Resource Name
and Type
or if one or more such resources exist, each of these must be GC'd and their time of deletion
be more than 24 Hours Ago.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param location: Location of the resource. Required.
:type location: str
:param input: Contains information about Resource type and Resource name. Required.
:type input: ~azure.mgmt.recoveryservices.models.CheckNameAvailabilityParameters
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: CheckNameAvailabilityResult or the result of cls(response)
:rtype: ~azure.mgmt.recoveryservices.models.CheckNameAvailabilityResult
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def check_name_availability(
self,
resource_group_name: str,
location: str,
input: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.CheckNameAvailabilityResult:
"""API to check for resource name availability.
A name is available if no other resource exists that has the same SubscriptionId, Resource Name
and Type
or if one or more such resources exist, each of these must be GC'd and their time of deletion
be more than 24 Hours Ago.
API to check for resource name availability.
A name is available if no other resource exists that has the same SubscriptionId, Resource Name
and Type
or if one or more such resources exist, each of these must be GC'd and their time of deletion
be more than 24 Hours Ago.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param location: Location of the resource. Required.
:type location: str
:param input: Contains information about Resource type and Resource name. Required.
:type input: IO[bytes]
:keyword content_type: Body Parameter content-type. Content type parameter for binary body.
Default value is "application/json".
:paramtype content_type: str
:return: CheckNameAvailabilityResult or the result of cls(response)
:rtype: ~azure.mgmt.recoveryservices.models.CheckNameAvailabilityResult
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs] @distributed_trace
def check_name_availability(
self,
resource_group_name: str,
location: str,
input: Union[_models.CheckNameAvailabilityParameters, IO[bytes]],
**kwargs: Any
) -> _models.CheckNameAvailabilityResult:
"""API to check for resource name availability.
A name is available if no other resource exists that has the same SubscriptionId, Resource Name
and Type
or if one or more such resources exist, each of these must be GC'd and their time of deletion
be more than 24 Hours Ago.
API to check for resource name availability.
A name is available if no other resource exists that has the same SubscriptionId, Resource Name
and Type
or if one or more such resources exist, each of these must be GC'd and their time of deletion
be more than 24 Hours Ago.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param location: Location of the resource. Required.
:type location: str
:param input: Contains information about Resource type and Resource name. Is either a
CheckNameAvailabilityParameters type or a IO[bytes] type. Required.
:type input: ~azure.mgmt.recoveryservices.models.CheckNameAvailabilityParameters or IO[bytes]
:return: CheckNameAvailabilityResult or the result of cls(response)
:rtype: ~azure.mgmt.recoveryservices.models.CheckNameAvailabilityResult
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.CheckNameAvailabilityResult] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_json = None
_content = None
if isinstance(input, (IOBase, bytes)):
_content = input
else:
_json = self._serialize.body(input, "CheckNameAvailabilityParameters")
_request = build_check_name_availability_request(
resource_group_name=resource_group_name,
location=location,
subscription_id=self._config.subscription_id,
api_version=api_version,
content_type=content_type,
json=_json,
content=_content,
headers=_headers,
params=_params,
)
_request = _convert_request(_request)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response, error_format=ARMErrorFormat)
deserialized = self._deserialize("CheckNameAvailabilityResult", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
@overload
def capabilities(
self,
location: str,
input: _models.ResourceCapabilities,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.CapabilitiesResponse:
"""API to get details about capabilities provided by Microsoft.RecoveryServices RP.
API to get details about capabilities provided by Microsoft.RecoveryServices RP.
:param location: Location of the resource. Required.
:type location: str
:param input: Contains information about Resource type and properties to get capabilities.
Required.
:type input: ~azure.mgmt.recoveryservices.models.ResourceCapabilities
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: CapabilitiesResponse or the result of cls(response)
:rtype: ~azure.mgmt.recoveryservices.models.CapabilitiesResponse
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def capabilities(
self, location: str, input: IO[bytes], *, content_type: str = "application/json", **kwargs: Any
) -> _models.CapabilitiesResponse:
"""API to get details about capabilities provided by Microsoft.RecoveryServices RP.
API to get details about capabilities provided by Microsoft.RecoveryServices RP.
:param location: Location of the resource. Required.
:type location: str
:param input: Contains information about Resource type and properties to get capabilities.
Required.
:type input: IO[bytes]
:keyword content_type: Body Parameter content-type. Content type parameter for binary body.
Default value is "application/json".
:paramtype content_type: str
:return: CapabilitiesResponse or the result of cls(response)
:rtype: ~azure.mgmt.recoveryservices.models.CapabilitiesResponse
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs] @distributed_trace
def capabilities(
self, location: str, input: Union[_models.ResourceCapabilities, IO[bytes]], **kwargs: Any
) -> _models.CapabilitiesResponse:
"""API to get details about capabilities provided by Microsoft.RecoveryServices RP.
API to get details about capabilities provided by Microsoft.RecoveryServices RP.
:param location: Location of the resource. Required.
:type location: str
:param input: Contains information about Resource type and properties to get capabilities. Is
either a ResourceCapabilities type or a IO[bytes] type. Required.
:type input: ~azure.mgmt.recoveryservices.models.ResourceCapabilities or IO[bytes]
:return: CapabilitiesResponse or the result of cls(response)
:rtype: ~azure.mgmt.recoveryservices.models.CapabilitiesResponse
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._config.api_version))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.CapabilitiesResponse] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_json = None
_content = None
if isinstance(input, (IOBase, bytes)):
_content = input
else:
_json = self._serialize.body(input, "ResourceCapabilities")
_request = build_capabilities_request(
location=location,
subscription_id=self._config.subscription_id,
api_version=api_version,
content_type=content_type,
json=_json,
content=_content,
headers=_headers,
params=_params,
)
_request = _convert_request(_request)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
raise HttpResponseError(response=response, error_format=ARMErrorFormat)
deserialized = self._deserialize("CapabilitiesResponse", pipeline_response)
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore