# pylint: disable=line-too-long,useless-suppression,too-many-lines
# 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 collections.abc import MutableMapping
from io import IOBase
import json
from typing import Any, Callable, Dict, IO, Iterable, Iterator, List, Optional, TypeVar, Union, cast, overload
import urllib.parse
from azure.core import PipelineClient
from azure.core.exceptions import (
ClientAuthenticationError,
HttpResponseError,
ResourceExistsError,
ResourceNotFoundError,
ResourceNotModifiedError,
StreamClosedError,
StreamConsumedError,
map_error,
)
from azure.core.paging import ItemPaged
from azure.core.pipeline import PipelineResponse
from azure.core.polling import LROPoller, NoPolling, PollingMethod
from azure.core.rest import HttpRequest, HttpResponse
from azure.core.tracing.decorator import distributed_trace
from azure.core.utils import case_insensitive_dict
from azure.mgmt.core.exceptions import ARMErrorFormat
from azure.mgmt.core.polling.arm_polling import ARMPolling
from .. import models as _models
from .._configuration import StandbyPoolMgmtClientConfiguration
from .._model_base import SdkJSONEncoder, _deserialize, _failsafe_deserialize
from .._serialization import Deserializer, Serializer
from .._validation import api_version_validation
T = TypeVar("T")
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
JSON = MutableMapping[str, Any]
_SERIALIZER = Serializer()
_SERIALIZER.client_side_validation = False
def build_operations_list_request(**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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/providers/Microsoft.StandbyPool/operations"
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machine_pools_get_request( # pylint: disable=name-too-long
resource_group_name: str, standby_virtual_machine_pool_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools/{standbyVirtualMachinePoolName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyVirtualMachinePoolName": _SERIALIZER.url(
"standby_virtual_machine_pool_name", standby_virtual_machine_pool_name, "str"
),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machine_pools_create_or_update_request( # pylint: disable=name-too-long
resource_group_name: str, standby_virtual_machine_pool_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools/{standbyVirtualMachinePoolName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyVirtualMachinePoolName": _SERIALIZER.url(
"standby_virtual_machine_pool_name", standby_virtual_machine_pool_name, "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="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machine_pools_delete_request( # pylint: disable=name-too-long
resource_group_name: str, standby_virtual_machine_pool_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools/{standbyVirtualMachinePoolName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyVirtualMachinePoolName": _SERIALIZER.url(
"standby_virtual_machine_pool_name", standby_virtual_machine_pool_name, "str"
),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machine_pools_update_request( # pylint: disable=name-too-long
resource_group_name: str, standby_virtual_machine_pool_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools/{standbyVirtualMachinePoolName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyVirtualMachinePoolName": _SERIALIZER.url(
"standby_virtual_machine_pool_name", standby_virtual_machine_pool_name, "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="PATCH", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machine_pools_list_by_resource_group_request( # pylint: disable=name-too-long
resource_group_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machine_pools_list_by_subscription_request( # pylint: disable=name-too-long
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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machines_get_request( # pylint: disable=name-too-long
resource_group_name: str,
standby_virtual_machine_pool_name: str,
standby_virtual_machine_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools/{standbyVirtualMachinePoolName}/standbyVirtualMachines/{standbyVirtualMachineName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyVirtualMachinePoolName": _SERIALIZER.url(
"standby_virtual_machine_pool_name", standby_virtual_machine_pool_name, "str"
),
"standbyVirtualMachineName": _SERIALIZER.url(
"standby_virtual_machine_name", standby_virtual_machine_name, "str"
),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machines_list_by_standby_virtual_machine_pool_resource_request( # pylint: disable=name-too-long
resource_group_name: str, standby_virtual_machine_pool_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools/{standbyVirtualMachinePoolName}/standbyVirtualMachines"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyVirtualMachinePoolName": _SERIALIZER.url(
"standby_virtual_machine_pool_name", standby_virtual_machine_pool_name, "str"
),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machine_pool_runtime_views_get_request( # pylint: disable=name-too-long
resource_group_name: str,
standby_virtual_machine_pool_name: str,
runtime_view: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools/{standbyVirtualMachinePoolName}/runtimeViews/{runtimeView}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyVirtualMachinePoolName": _SERIALIZER.url(
"standby_virtual_machine_pool_name", standby_virtual_machine_pool_name, "str"
),
"runtimeView": _SERIALIZER.url("runtime_view", runtime_view, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_virtual_machine_pool_runtime_views_list_by_standby_pool_request( # pylint: disable=name-too-long
resource_group_name: str, standby_virtual_machine_pool_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyVirtualMachinePools/{standbyVirtualMachinePoolName}/runtimeViews"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyVirtualMachinePoolName": _SERIALIZER.url(
"standby_virtual_machine_pool_name", standby_virtual_machine_pool_name, "str"
),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_container_group_pools_get_request( # pylint: disable=name-too-long
resource_group_name: str, standby_container_group_pool_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyContainerGroupPools/{standbyContainerGroupPoolName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyContainerGroupPoolName": _SERIALIZER.url(
"standby_container_group_pool_name", standby_container_group_pool_name, "str"
),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_container_group_pools_create_or_update_request( # pylint: disable=name-too-long
resource_group_name: str, standby_container_group_pool_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyContainerGroupPools/{standbyContainerGroupPoolName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyContainerGroupPoolName": _SERIALIZER.url(
"standby_container_group_pool_name", standby_container_group_pool_name, "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="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_container_group_pools_delete_request( # pylint: disable=name-too-long
resource_group_name: str, standby_container_group_pool_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyContainerGroupPools/{standbyContainerGroupPoolName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyContainerGroupPoolName": _SERIALIZER.url(
"standby_container_group_pool_name", standby_container_group_pool_name, "str"
),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_container_group_pools_update_request( # pylint: disable=name-too-long
resource_group_name: str, standby_container_group_pool_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyContainerGroupPools/{standbyContainerGroupPoolName}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyContainerGroupPoolName": _SERIALIZER.url(
"standby_container_group_pool_name", standby_container_group_pool_name, "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="PATCH", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_container_group_pools_list_by_resource_group_request( # pylint: disable=name-too-long
resource_group_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyContainerGroupPools"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_container_group_pools_list_by_subscription_request( # pylint: disable=name-too-long
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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/providers/Microsoft.StandbyPool/standbyContainerGroupPools"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_container_group_pool_runtime_views_get_request( # pylint: disable=name-too-long
resource_group_name: str,
standby_container_group_pool_name: str,
runtime_view: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyContainerGroupPools/{standbyContainerGroupPoolName}/runtimeViews/{runtimeView}"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyContainerGroupPoolName": _SERIALIZER.url(
"standby_container_group_pool_name", standby_container_group_pool_name, "str"
),
"runtimeView": _SERIALIZER.url("runtime_view", runtime_view, "str"),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_standby_container_group_pool_runtime_views_list_by_standby_pool_request( # pylint: disable=name-too-long
resource_group_name: str, standby_container_group_pool_name: 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", "2025-03-01"))
accept = _headers.pop("Accept", "application/json")
# Construct URL
_url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.StandbyPool/standbyContainerGroupPools/{standbyContainerGroupPoolName}/runtimeViews"
path_format_arguments = {
"subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
"resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
"standbyContainerGroupPoolName": _SERIALIZER.url(
"standby_container_group_pool_name", standby_container_group_pool_name, "str"
),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")
# Construct headers
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
[docs]
class Operations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.standbypool.StandbyPoolMgmtClient`'s
:attr:`operations` attribute.
"""
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: StandbyPoolMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
def list(self, **kwargs: Any) -> Iterable["_models.Operation"]:
"""List the operations for the provider.
:return: An iterator like instance of Operation
:rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.standbypool.models.Operation]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.Operation]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_operations_list_request(
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.Operation], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_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)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
class StandbyVirtualMachinePoolsOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.standbypool.StandbyPoolMgmtClient`'s
:attr:`standby_virtual_machine_pools` attribute.
"""
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: StandbyPoolMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
def get(
self, resource_group_name: str, standby_virtual_machine_pool_name: str, **kwargs: Any
) -> _models.StandbyVirtualMachinePoolResource:
"""Get a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:return: StandbyVirtualMachinePoolResource. The StandbyVirtualMachinePoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[_models.StandbyVirtualMachinePoolResource] = kwargs.pop("cls", None)
_request = build_standby_virtual_machine_pools_get_request(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.StandbyVirtualMachinePoolResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
def _create_or_update_initial(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
resource: Union[_models.StandbyVirtualMachinePoolResource, JSON, IO[bytes]],
**kwargs: Any
) -> Iterator[bytes]:
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(resource, (IOBase, bytes)):
_content = resource
else:
_content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_standby_virtual_machine_pools_create_or_update_request(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
subscription_id=self._config.subscription_id,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = True
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, 201]:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
response_headers = {}
if response.status_code == 201:
response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After"))
deserialized = response.iter_bytes()
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@overload
def begin_create_or_update(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
resource: _models.StandbyVirtualMachinePoolResource,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.StandbyVirtualMachinePoolResource]:
"""Create a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param resource: Resource create parameters. Required.
:type resource: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: An instance of LROPoller that returns StandbyVirtualMachinePoolResource. The
StandbyVirtualMachinePoolResource is compatible with MutableMapping
:rtype:
~azure.core.polling.LROPoller[~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
resource: JSON,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.StandbyVirtualMachinePoolResource]:
"""Create a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param resource: Resource create parameters. Required.
:type resource: JSON
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: An instance of LROPoller that returns StandbyVirtualMachinePoolResource. The
StandbyVirtualMachinePoolResource is compatible with MutableMapping
:rtype:
~azure.core.polling.LROPoller[~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
resource: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.StandbyVirtualMachinePoolResource]:
"""Create a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param resource: Resource create parameters. Required.
:type resource: 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: An instance of LROPoller that returns StandbyVirtualMachinePoolResource. The
StandbyVirtualMachinePoolResource is compatible with MutableMapping
:rtype:
~azure.core.polling.LROPoller[~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
def begin_create_or_update(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
resource: Union[_models.StandbyVirtualMachinePoolResource, JSON, IO[bytes]],
**kwargs: Any
) -> LROPoller[_models.StandbyVirtualMachinePoolResource]:
"""Create a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param resource: Resource create parameters. Is one of the following types:
StandbyVirtualMachinePoolResource, JSON, IO[bytes] Required.
:type resource: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource or JSON or
IO[bytes]
:return: An instance of LROPoller that returns StandbyVirtualMachinePoolResource. The
StandbyVirtualMachinePoolResource is compatible with MutableMapping
:rtype:
~azure.core.polling.LROPoller[~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.StandbyVirtualMachinePoolResource] = kwargs.pop("cls", None)
polling: Union[bool, PollingMethod] = kwargs.pop("polling", True)
lro_delay = kwargs.pop("polling_interval", self._config.polling_interval)
cont_token: Optional[str] = kwargs.pop("continuation_token", None)
if cont_token is None:
raw_result = self._create_or_update_initial(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
resource=resource,
content_type=content_type,
cls=lambda x, y, z: x,
headers=_headers,
params=_params,
**kwargs
)
raw_result.http_response.read() # type: ignore
kwargs.pop("error_map", None)
def get_long_running_output(pipeline_response):
response = pipeline_response.http_response
deserialized = _deserialize(_models.StandbyVirtualMachinePoolResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
if polling is True:
polling_method: PollingMethod = cast(
PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
)
elif polling is False:
polling_method = cast(PollingMethod, NoPolling())
else:
polling_method = polling
if cont_token:
return LROPoller[_models.StandbyVirtualMachinePoolResource].from_continuation_token(
polling_method=polling_method,
continuation_token=cont_token,
client=self._client,
deserialization_callback=get_long_running_output,
)
return LROPoller[_models.StandbyVirtualMachinePoolResource](
self._client, raw_result, get_long_running_output, polling_method # type: ignore
)
def _delete_initial(
self, resource_group_name: str, standby_virtual_machine_pool_name: str, **kwargs: Any
) -> Iterator[bytes]:
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None)
_request = build_standby_virtual_machine_pools_delete_request(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = True
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 [202, 204]:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
response_headers = {}
if response.status_code == 202:
response_headers["Location"] = self._deserialize("str", response.headers.get("Location"))
response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After"))
deserialized = response.iter_bytes()
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
def begin_delete(
self, resource_group_name: str, standby_virtual_machine_pool_name: str, **kwargs: Any
) -> LROPoller[None]:
"""Delete a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:return: An instance of LROPoller that returns None
:rtype: ~azure.core.polling.LROPoller[None]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[None] = kwargs.pop("cls", None)
polling: Union[bool, PollingMethod] = kwargs.pop("polling", True)
lro_delay = kwargs.pop("polling_interval", self._config.polling_interval)
cont_token: Optional[str] = kwargs.pop("continuation_token", None)
if cont_token is None:
raw_result = self._delete_initial(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
cls=lambda x, y, z: x,
headers=_headers,
params=_params,
**kwargs
)
raw_result.http_response.read() # type: ignore
kwargs.pop("error_map", None)
def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements
if cls:
return cls(pipeline_response, None, {}) # type: ignore
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
if polling is True:
polling_method: PollingMethod = cast(
PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
)
elif polling is False:
polling_method = cast(PollingMethod, NoPolling())
else:
polling_method = polling
if cont_token:
return LROPoller[None].from_continuation_token(
polling_method=polling_method,
continuation_token=cont_token,
client=self._client,
deserialization_callback=get_long_running_output,
)
return LROPoller[None](self._client, raw_result, get_long_running_output, polling_method) # type: ignore
@overload
def update(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
properties: _models.StandbyVirtualMachinePoolResourceUpdate,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.StandbyVirtualMachinePoolResource:
"""Update a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param properties: The resource properties to be updated. Required.
:type properties: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResourceUpdate
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: StandbyVirtualMachinePoolResource. The StandbyVirtualMachinePoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
properties: JSON,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.StandbyVirtualMachinePoolResource:
"""Update a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param properties: The resource properties to be updated. Required.
:type properties: JSON
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: StandbyVirtualMachinePoolResource. The StandbyVirtualMachinePoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
properties: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.StandbyVirtualMachinePoolResource:
"""Update a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param properties: The resource properties to be updated. Required.
:type properties: 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: StandbyVirtualMachinePoolResource. The StandbyVirtualMachinePoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
def update(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
properties: Union[_models.StandbyVirtualMachinePoolResourceUpdate, JSON, IO[bytes]],
**kwargs: Any
) -> _models.StandbyVirtualMachinePoolResource:
"""Update a StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param properties: The resource properties to be updated. Is one of the following types:
StandbyVirtualMachinePoolResourceUpdate, JSON, IO[bytes] Required.
:type properties: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResourceUpdate or
JSON or IO[bytes]
:return: StandbyVirtualMachinePoolResource. The StandbyVirtualMachinePoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.StandbyVirtualMachinePoolResource] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(properties, (IOBase, bytes)):
_content = properties
else:
_content = json.dumps(properties, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_standby_virtual_machine_pools_update_request(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
subscription_id=self._config.subscription_id,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.StandbyVirtualMachinePoolResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
def list_by_resource_group(
self, resource_group_name: str, **kwargs: Any
) -> Iterable["_models.StandbyVirtualMachinePoolResource"]:
"""List StandbyVirtualMachinePoolResource resources by resource group.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:return: An iterator like instance of StandbyVirtualMachinePoolResource
:rtype:
~azure.core.paging.ItemPaged[~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.StandbyVirtualMachinePoolResource]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_standby_virtual_machine_pools_list_by_resource_group_request(
resource_group_name=resource_group_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.StandbyVirtualMachinePoolResource], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_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)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
@distributed_trace
def list_by_subscription(self, **kwargs: Any) -> Iterable["_models.StandbyVirtualMachinePoolResource"]:
"""List StandbyVirtualMachinePoolResource resources by subscription ID.
:return: An iterator like instance of StandbyVirtualMachinePoolResource
:rtype:
~azure.core.paging.ItemPaged[~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.StandbyVirtualMachinePoolResource]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_standby_virtual_machine_pools_list_by_subscription_request(
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.StandbyVirtualMachinePoolResource], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_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)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
class StandbyVirtualMachinesOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.standbypool.StandbyPoolMgmtClient`'s
:attr:`standby_virtual_machines` attribute.
"""
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: StandbyPoolMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
def get(
self,
resource_group_name: str,
standby_virtual_machine_pool_name: str,
standby_virtual_machine_name: str,
**kwargs: Any
) -> _models.StandbyVirtualMachineResource:
"""Get a StandbyVirtualMachineResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param standby_virtual_machine_name: Name of the standby virtual machine. Required.
:type standby_virtual_machine_name: str
:return: StandbyVirtualMachineResource. The StandbyVirtualMachineResource is compatible with
MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyVirtualMachineResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[_models.StandbyVirtualMachineResource] = kwargs.pop("cls", None)
_request = build_standby_virtual_machines_get_request(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
standby_virtual_machine_name=standby_virtual_machine_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.StandbyVirtualMachineResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
def list_by_standby_virtual_machine_pool_resource( # pylint: disable=name-too-long
self, resource_group_name: str, standby_virtual_machine_pool_name: str, **kwargs: Any
) -> Iterable["_models.StandbyVirtualMachineResource"]:
"""List StandbyVirtualMachineResource resources by StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:return: An iterator like instance of StandbyVirtualMachineResource
:rtype:
~azure.core.paging.ItemPaged[~azure.mgmt.standbypool.models.StandbyVirtualMachineResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.StandbyVirtualMachineResource]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_standby_virtual_machines_list_by_standby_virtual_machine_pool_resource_request(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.StandbyVirtualMachineResource], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_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)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
class StandbyVirtualMachinePoolRuntimeViewsOperations: # pylint: disable=name-too-long
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.standbypool.StandbyPoolMgmtClient`'s
:attr:`standby_virtual_machine_pool_runtime_views` attribute.
"""
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: StandbyPoolMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2024-03-01-preview",
params_added_on={
"2024-03-01-preview": [
"api_version",
"subscription_id",
"resource_group_name",
"standby_virtual_machine_pool_name",
"runtime_view",
"accept",
]
},
)
def get(
self, resource_group_name: str, standby_virtual_machine_pool_name: str, runtime_view: str, **kwargs: Any
) -> _models.StandbyVirtualMachinePoolRuntimeViewResource:
"""Get a StandbyVirtualMachinePoolRuntimeViewResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:param runtime_view: The unique identifier for the runtime view. The input string should be the
word 'latest', which will get the latest runtime view of the pool, otherwise the request will
fail with NotFound exception. Required.
:type runtime_view: str
:return: StandbyVirtualMachinePoolRuntimeViewResource. The
StandbyVirtualMachinePoolRuntimeViewResource is compatible with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolRuntimeViewResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[_models.StandbyVirtualMachinePoolRuntimeViewResource] = kwargs.pop("cls", None)
_request = build_standby_virtual_machine_pool_runtime_views_get_request(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
runtime_view=runtime_view,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.StandbyVirtualMachinePoolRuntimeViewResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2024-03-01-preview",
params_added_on={
"2024-03-01-preview": [
"api_version",
"subscription_id",
"resource_group_name",
"standby_virtual_machine_pool_name",
"accept",
]
},
)
def list_by_standby_pool(
self, resource_group_name: str, standby_virtual_machine_pool_name: str, **kwargs: Any
) -> Iterable["_models.StandbyVirtualMachinePoolRuntimeViewResource"]:
"""List StandbyVirtualMachinePoolRuntimeViewResource resources by
StandbyVirtualMachinePoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_virtual_machine_pool_name: Name of the standby virtual machine pool. Required.
:type standby_virtual_machine_pool_name: str
:return: An iterator like instance of StandbyVirtualMachinePoolRuntimeViewResource
:rtype:
~azure.core.paging.ItemPaged[~azure.mgmt.standbypool.models.StandbyVirtualMachinePoolRuntimeViewResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.StandbyVirtualMachinePoolRuntimeViewResource]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_standby_virtual_machine_pool_runtime_views_list_by_standby_pool_request(
resource_group_name=resource_group_name,
standby_virtual_machine_pool_name=standby_virtual_machine_pool_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(
List[_models.StandbyVirtualMachinePoolRuntimeViewResource], deserialized.get("value", [])
)
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_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)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
class StandbyContainerGroupPoolsOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.standbypool.StandbyPoolMgmtClient`'s
:attr:`standby_container_group_pools` attribute.
"""
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: StandbyPoolMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
def get(
self, resource_group_name: str, standby_container_group_pool_name: str, **kwargs: Any
) -> _models.StandbyContainerGroupPoolResource:
"""Get a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:return: StandbyContainerGroupPoolResource. The StandbyContainerGroupPoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[_models.StandbyContainerGroupPoolResource] = kwargs.pop("cls", None)
_request = build_standby_container_group_pools_get_request(
resource_group_name=resource_group_name,
standby_container_group_pool_name=standby_container_group_pool_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.StandbyContainerGroupPoolResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
def _create_or_update_initial(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
resource: Union[_models.StandbyContainerGroupPoolResource, JSON, IO[bytes]],
**kwargs: Any
) -> Iterator[bytes]:
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(resource, (IOBase, bytes)):
_content = resource
else:
_content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_standby_container_group_pools_create_or_update_request(
resource_group_name=resource_group_name,
standby_container_group_pool_name=standby_container_group_pool_name,
subscription_id=self._config.subscription_id,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = True
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, 201]:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
response_headers = {}
if response.status_code == 201:
response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After"))
deserialized = response.iter_bytes()
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@overload
def begin_create_or_update(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
resource: _models.StandbyContainerGroupPoolResource,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.StandbyContainerGroupPoolResource]:
"""Create a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param resource: Resource create parameters. Required.
:type resource: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: An instance of LROPoller that returns StandbyContainerGroupPoolResource. The
StandbyContainerGroupPoolResource is compatible with MutableMapping
:rtype:
~azure.core.polling.LROPoller[~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
resource: JSON,
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.StandbyContainerGroupPoolResource]:
"""Create a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param resource: Resource create parameters. Required.
:type resource: JSON
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: An instance of LROPoller that returns StandbyContainerGroupPoolResource. The
StandbyContainerGroupPoolResource is compatible with MutableMapping
:rtype:
~azure.core.polling.LROPoller[~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def begin_create_or_update(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
resource: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> LROPoller[_models.StandbyContainerGroupPoolResource]:
"""Create a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param resource: Resource create parameters. Required.
:type resource: 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: An instance of LROPoller that returns StandbyContainerGroupPoolResource. The
StandbyContainerGroupPoolResource is compatible with MutableMapping
:rtype:
~azure.core.polling.LROPoller[~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
def begin_create_or_update(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
resource: Union[_models.StandbyContainerGroupPoolResource, JSON, IO[bytes]],
**kwargs: Any
) -> LROPoller[_models.StandbyContainerGroupPoolResource]:
"""Create a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param resource: Resource create parameters. Is one of the following types:
StandbyContainerGroupPoolResource, JSON, IO[bytes] Required.
:type resource: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource or JSON or
IO[bytes]
:return: An instance of LROPoller that returns StandbyContainerGroupPoolResource. The
StandbyContainerGroupPoolResource is compatible with MutableMapping
:rtype:
~azure.core.polling.LROPoller[~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.StandbyContainerGroupPoolResource] = kwargs.pop("cls", None)
polling: Union[bool, PollingMethod] = kwargs.pop("polling", True)
lro_delay = kwargs.pop("polling_interval", self._config.polling_interval)
cont_token: Optional[str] = kwargs.pop("continuation_token", None)
if cont_token is None:
raw_result = self._create_or_update_initial(
resource_group_name=resource_group_name,
standby_container_group_pool_name=standby_container_group_pool_name,
resource=resource,
content_type=content_type,
cls=lambda x, y, z: x,
headers=_headers,
params=_params,
**kwargs
)
raw_result.http_response.read() # type: ignore
kwargs.pop("error_map", None)
def get_long_running_output(pipeline_response):
response = pipeline_response.http_response
deserialized = _deserialize(_models.StandbyContainerGroupPoolResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
if polling is True:
polling_method: PollingMethod = cast(
PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
)
elif polling is False:
polling_method = cast(PollingMethod, NoPolling())
else:
polling_method = polling
if cont_token:
return LROPoller[_models.StandbyContainerGroupPoolResource].from_continuation_token(
polling_method=polling_method,
continuation_token=cont_token,
client=self._client,
deserialization_callback=get_long_running_output,
)
return LROPoller[_models.StandbyContainerGroupPoolResource](
self._client, raw_result, get_long_running_output, polling_method # type: ignore
)
def _delete_initial(
self, resource_group_name: str, standby_container_group_pool_name: str, **kwargs: Any
) -> Iterator[bytes]:
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None)
_request = build_standby_container_group_pools_delete_request(
resource_group_name=resource_group_name,
standby_container_group_pool_name=standby_container_group_pool_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = True
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 [202, 204]:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
response_headers = {}
if response.status_code == 202:
response_headers["Location"] = self._deserialize("str", response.headers.get("Location"))
response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After"))
deserialized = response.iter_bytes()
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
def begin_delete(
self, resource_group_name: str, standby_container_group_pool_name: str, **kwargs: Any
) -> LROPoller[None]:
"""Delete a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:return: An instance of LROPoller that returns None
:rtype: ~azure.core.polling.LROPoller[None]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[None] = kwargs.pop("cls", None)
polling: Union[bool, PollingMethod] = kwargs.pop("polling", True)
lro_delay = kwargs.pop("polling_interval", self._config.polling_interval)
cont_token: Optional[str] = kwargs.pop("continuation_token", None)
if cont_token is None:
raw_result = self._delete_initial(
resource_group_name=resource_group_name,
standby_container_group_pool_name=standby_container_group_pool_name,
cls=lambda x, y, z: x,
headers=_headers,
params=_params,
**kwargs
)
raw_result.http_response.read() # type: ignore
kwargs.pop("error_map", None)
def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements
if cls:
return cls(pipeline_response, None, {}) # type: ignore
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
if polling is True:
polling_method: PollingMethod = cast(
PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
)
elif polling is False:
polling_method = cast(PollingMethod, NoPolling())
else:
polling_method = polling
if cont_token:
return LROPoller[None].from_continuation_token(
polling_method=polling_method,
continuation_token=cont_token,
client=self._client,
deserialization_callback=get_long_running_output,
)
return LROPoller[None](self._client, raw_result, get_long_running_output, polling_method) # type: ignore
@overload
def update(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
properties: _models.StandbyContainerGroupPoolResourceUpdate,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.StandbyContainerGroupPoolResource:
"""Update a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param properties: The resource properties to be updated. Required.
:type properties: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResourceUpdate
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: StandbyContainerGroupPoolResource. The StandbyContainerGroupPoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
properties: JSON,
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.StandbyContainerGroupPoolResource:
"""Update a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param properties: The resource properties to be updated. Required.
:type properties: JSON
:keyword content_type: Body Parameter content-type. Content type parameter for JSON body.
Default value is "application/json".
:paramtype content_type: str
:return: StandbyContainerGroupPoolResource. The StandbyContainerGroupPoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
@overload
def update(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
properties: IO[bytes],
*,
content_type: str = "application/json",
**kwargs: Any
) -> _models.StandbyContainerGroupPoolResource:
"""Update a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param properties: The resource properties to be updated. Required.
:type properties: 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: StandbyContainerGroupPoolResource. The StandbyContainerGroupPoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
[docs]
@distributed_trace
def update(
self,
resource_group_name: str,
standby_container_group_pool_name: str,
properties: Union[_models.StandbyContainerGroupPoolResourceUpdate, JSON, IO[bytes]],
**kwargs: Any
) -> _models.StandbyContainerGroupPoolResource:
"""Update a StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param properties: The resource properties to be updated. Is one of the following types:
StandbyContainerGroupPoolResourceUpdate, JSON, IO[bytes] Required.
:type properties: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResourceUpdate or
JSON or IO[bytes]
:return: StandbyContainerGroupPoolResource. The StandbyContainerGroupPoolResource is compatible
with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
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 = kwargs.pop("params", {}) or {}
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
cls: ClsType[_models.StandbyContainerGroupPoolResource] = kwargs.pop("cls", None)
content_type = content_type or "application/json"
_content = None
if isinstance(properties, (IOBase, bytes)):
_content = properties
else:
_content = json.dumps(properties, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore
_request = build_standby_container_group_pools_update_request(
resource_group_name=resource_group_name,
standby_container_group_pool_name=standby_container_group_pool_name,
subscription_id=self._config.subscription_id,
content_type=content_type,
api_version=self._config.api_version,
content=_content,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.StandbyContainerGroupPoolResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
def list_by_resource_group(
self, resource_group_name: str, **kwargs: Any
) -> Iterable["_models.StandbyContainerGroupPoolResource"]:
"""List StandbyContainerGroupPoolResource resources by resource group.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:return: An iterator like instance of StandbyContainerGroupPoolResource
:rtype:
~azure.core.paging.ItemPaged[~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.StandbyContainerGroupPoolResource]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_standby_container_group_pools_list_by_resource_group_request(
resource_group_name=resource_group_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.StandbyContainerGroupPoolResource], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_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)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
@distributed_trace
def list_by_subscription(self, **kwargs: Any) -> Iterable["_models.StandbyContainerGroupPoolResource"]:
"""List StandbyContainerGroupPoolResource resources by subscription ID.
:return: An iterator like instance of StandbyContainerGroupPoolResource
:rtype:
~azure.core.paging.ItemPaged[~azure.mgmt.standbypool.models.StandbyContainerGroupPoolResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.StandbyContainerGroupPoolResource]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_standby_container_group_pools_list_by_subscription_request(
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(List[_models.StandbyContainerGroupPoolResource], deserialized.get("value", []))
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_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)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)
[docs]
class StandbyContainerGroupPoolRuntimeViewsOperations: # pylint: disable=name-too-long
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.mgmt.standbypool.StandbyPoolMgmtClient`'s
:attr:`standby_container_group_pool_runtime_views` attribute.
"""
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: StandbyPoolMgmtClientConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2024-03-01-preview",
params_added_on={
"2024-03-01-preview": [
"api_version",
"subscription_id",
"resource_group_name",
"standby_container_group_pool_name",
"runtime_view",
"accept",
]
},
)
def get(
self, resource_group_name: str, standby_container_group_pool_name: str, runtime_view: str, **kwargs: Any
) -> _models.StandbyContainerGroupPoolRuntimeViewResource:
"""Get a StandbyContainerGroupPoolRuntimeViewResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:param runtime_view: The unique identifier for the runtime view. The input string should be the
word 'latest', which will get the latest runtime view of the pool, otherwise the request will
fail with NotFound exception. Required.
:type runtime_view: str
:return: StandbyContainerGroupPoolRuntimeViewResource. The
StandbyContainerGroupPoolRuntimeViewResource is compatible with MutableMapping
:rtype: ~azure.mgmt.standbypool.models.StandbyContainerGroupPoolRuntimeViewResource
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[_models.StandbyContainerGroupPoolRuntimeViewResource] = kwargs.pop("cls", None)
_request = build_standby_container_group_pool_runtime_views_get_request(
resource_group_name=resource_group_name,
standby_container_group_pool_name=standby_container_group_pool_name,
runtime_view=runtime_view,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
_stream = kwargs.pop("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]:
if _stream:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
if _stream:
deserialized = response.iter_bytes()
else:
deserialized = _deserialize(_models.StandbyContainerGroupPoolRuntimeViewResource, response.json())
if cls:
return cls(pipeline_response, deserialized, {}) # type: ignore
return deserialized # type: ignore
[docs]
@distributed_trace
@api_version_validation(
method_added_on="2024-03-01-preview",
params_added_on={
"2024-03-01-preview": [
"api_version",
"subscription_id",
"resource_group_name",
"standby_container_group_pool_name",
"accept",
]
},
)
def list_by_standby_pool(
self, resource_group_name: str, standby_container_group_pool_name: str, **kwargs: Any
) -> Iterable["_models.StandbyContainerGroupPoolRuntimeViewResource"]:
"""List StandbyContainerGroupPoolRuntimeViewResource resources by
StandbyContainerGroupPoolResource.
:param resource_group_name: The name of the resource group. The name is case insensitive.
Required.
:type resource_group_name: str
:param standby_container_group_pool_name: Name of the standby container group pool. Required.
:type standby_container_group_pool_name: str
:return: An iterator like instance of StandbyContainerGroupPoolRuntimeViewResource
:rtype:
~azure.core.paging.ItemPaged[~azure.mgmt.standbypool.models.StandbyContainerGroupPoolRuntimeViewResource]
:raises ~azure.core.exceptions.HttpResponseError:
"""
_headers = kwargs.pop("headers", {}) or {}
_params = kwargs.pop("params", {}) or {}
cls: ClsType[List[_models.StandbyContainerGroupPoolRuntimeViewResource]] = kwargs.pop("cls", None)
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
def prepare_request(next_link=None):
if not next_link:
_request = build_standby_container_group_pool_runtime_views_list_by_standby_pool_request(
resource_group_name=resource_group_name,
standby_container_group_pool_name=standby_container_group_pool_name,
subscription_id=self._config.subscription_id,
api_version=self._config.api_version,
headers=_headers,
params=_params,
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
else:
# make call to next link with the client's api-version
_parsed_next_link = urllib.parse.urlparse(next_link)
_next_request_params = case_insensitive_dict(
{
key: [urllib.parse.quote(v) for v in value]
for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items()
}
)
_next_request_params["api-version"] = self._config.api_version
_request = HttpRequest(
"GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params
)
path_format_arguments = {
"endpoint": self._serialize.url(
"self._config.base_url", self._config.base_url, "str", skip_quote=True
),
}
_request.url = self._client.format_url(_request.url, **path_format_arguments)
return _request
def extract_data(pipeline_response):
deserialized = pipeline_response.http_response.json()
list_of_elem = _deserialize(
List[_models.StandbyContainerGroupPoolRuntimeViewResource], deserialized.get("value", [])
)
if cls:
list_of_elem = cls(list_of_elem) # type: ignore
return deserialized.get("nextLink") or None, iter(list_of_elem)
def get_next(next_link=None):
_request = prepare_request(next_link)
_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)
error = _failsafe_deserialize(_models.ErrorResponse, response.json())
raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat)
return pipeline_response
return ItemPaged(get_next, extract_data)