Source code for azure.mgmt.compute.v2024_03_02.operations._operations

# pylint: disable=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) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
from io import IOBase
import sys
from typing import Any, Callable, Dict, IO, Iterable, Iterator, Optional, TypeVar, Union, cast, overload
import urllib.parse

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 ..._serialization import Serializer

if sys.version_info >= (3, 9):
    from collections.abc import MutableMapping
else:
    from typing import MutableMapping  # type: ignore
T = TypeVar("T")
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]

_SERIALIZER = Serializer()
_SERIALIZER.client_side_validation = False


def build_disks_create_or_update_request(
    resource_group_name: str, disk_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/disks/{diskName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskName": _SERIALIZER.url("disk_name", disk_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_disks_update_request(
    resource_group_name: str, disk_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/disks/{diskName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskName": _SERIALIZER.url("disk_name", disk_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_disks_get_request(
    resource_group_name: str, disk_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/disks/{diskName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskName": _SERIALIZER.url("disk_name", disk_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_disks_delete_request(
    resource_group_name: str, disk_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})

    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-03-02"))
    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/disks/{diskName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskName": _SERIALIZER.url("disk_name", disk_name, "str"),
    }

    _url: str = _url.format(**path_format_arguments)  # type: ignore

    # Construct parameters
    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")

    return HttpRequest(method="DELETE", url=_url, params=_params, **kwargs)


def build_disks_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/disks",
    )  # pylint: disable=line-too-long
    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_disks_list_request(subscription_id: str, **kwargs: Any) -> HttpRequest:
    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})

    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop("template_url", "/subscriptions/{subscriptionId}/providers/Microsoft.Compute/disks")
    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_disks_grant_access_request(
    resource_group_name: str, disk_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/disks/{diskName}/beginGetAccess",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskName": _SERIALIZER.url("disk_name", disk_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="POST", url=_url, params=_params, headers=_headers, **kwargs)


def build_disks_revoke_access_request(
    resource_group_name: str, disk_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})

    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-03-02"))
    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/disks/{diskName}/endGetAccess",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskName": _SERIALIZER.url("disk_name", disk_name, "str"),
    }

    _url: str = _url.format(**path_format_arguments)  # type: ignore

    # Construct parameters
    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")

    return HttpRequest(method="POST", url=_url, params=_params, **kwargs)


def build_disk_accesses_create_or_update_request(  # pylint: disable=name-too-long
    resource_group_name: str, disk_access_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_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_disk_accesses_update_request(
    resource_group_name: str, disk_access_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_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_disk_accesses_get_request(
    resource_group_name: str, disk_access_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_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_disk_accesses_delete_request(
    resource_group_name: str, disk_access_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_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_disk_accesses_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses",
    )  # pylint: disable=line-too-long
    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_disk_accesses_list_request(subscription_id: str, **kwargs: Any) -> HttpRequest:
    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})

    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop("template_url", "/subscriptions/{subscriptionId}/providers/Microsoft.Compute/diskAccesses")
    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_disk_accesses_get_private_link_resources_request(  # pylint: disable=name-too-long
    resource_group_name: str, disk_access_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}/privateLinkResources",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_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_disk_accesses_update_a_private_endpoint_connection_request(  # pylint: disable=name-too-long
    resource_group_name: str,
    disk_access_name: str,
    private_endpoint_connection_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}/privateEndpointConnections/{privateEndpointConnectionName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_name, "str"),
        "privateEndpointConnectionName": _SERIALIZER.url(
            "private_endpoint_connection_name", private_endpoint_connection_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_disk_accesses_get_a_private_endpoint_connection_request(  # pylint: disable=name-too-long
    resource_group_name: str,
    disk_access_name: str,
    private_endpoint_connection_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}/privateEndpointConnections/{privateEndpointConnectionName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_name, "str"),
        "privateEndpointConnectionName": _SERIALIZER.url(
            "private_endpoint_connection_name", private_endpoint_connection_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_disk_accesses_delete_a_private_endpoint_connection_request(  # pylint: disable=name-too-long
    resource_group_name: str,
    disk_access_name: str,
    private_endpoint_connection_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}/privateEndpointConnections/{privateEndpointConnectionName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_name, "str"),
        "privateEndpointConnectionName": _SERIALIZER.url(
            "private_endpoint_connection_name", private_endpoint_connection_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_disk_accesses_list_private_endpoint_connections_request(  # pylint: disable=name-too-long
    resource_group_name: str, disk_access_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskAccesses/{diskAccessName}/privateEndpointConnections",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskAccessName": _SERIALIZER.url("disk_access_name", disk_access_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_disk_encryption_sets_create_or_update_request(  # pylint: disable=name-too-long
    resource_group_name: str, disk_encryption_set_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskEncryptionSets/{diskEncryptionSetName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskEncryptionSetName": _SERIALIZER.url("disk_encryption_set_name", disk_encryption_set_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_disk_encryption_sets_update_request(  # pylint: disable=name-too-long
    resource_group_name: str, disk_encryption_set_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskEncryptionSets/{diskEncryptionSetName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskEncryptionSetName": _SERIALIZER.url("disk_encryption_set_name", disk_encryption_set_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_disk_encryption_sets_get_request(
    resource_group_name: str, disk_encryption_set_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskEncryptionSets/{diskEncryptionSetName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskEncryptionSetName": _SERIALIZER.url("disk_encryption_set_name", disk_encryption_set_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_disk_encryption_sets_delete_request(  # pylint: disable=name-too-long
    resource_group_name: str, disk_encryption_set_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskEncryptionSets/{diskEncryptionSetName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskEncryptionSetName": _SERIALIZER.url("disk_encryption_set_name", disk_encryption_set_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_disk_encryption_sets_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskEncryptionSets",
    )  # pylint: disable=line-too-long
    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_disk_encryption_sets_list_request(subscription_id: str, **kwargs: Any) -> HttpRequest:
    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})

    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop("template_url", "/subscriptions/{subscriptionId}/providers/Microsoft.Compute/diskEncryptionSets")
    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_disk_encryption_sets_list_associated_resources_request(  # pylint: disable=name-too-long
    resource_group_name: str, disk_encryption_set_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/diskEncryptionSets/{diskEncryptionSetName}/associatedResources",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "diskEncryptionSetName": _SERIALIZER.url("disk_encryption_set_name", disk_encryption_set_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_disk_restore_point_get_request(
    resource_group_name: str,
    restore_point_collection_name: str,
    vm_restore_point_name: str,
    disk_restore_point_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/restorePointCollections/{restorePointCollectionName}/restorePoints/{vmRestorePointName}/diskRestorePoints/{diskRestorePointName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "restorePointCollectionName": _SERIALIZER.url(
            "restore_point_collection_name", restore_point_collection_name, "str"
        ),
        "vmRestorePointName": _SERIALIZER.url("vm_restore_point_name", vm_restore_point_name, "str"),
        "diskRestorePointName": _SERIALIZER.url("disk_restore_point_name", disk_restore_point_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_disk_restore_point_list_by_restore_point_request(  # pylint: disable=name-too-long
    resource_group_name: str,
    restore_point_collection_name: str,
    vm_restore_point_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/restorePointCollections/{restorePointCollectionName}/restorePoints/{vmRestorePointName}/diskRestorePoints",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "restorePointCollectionName": _SERIALIZER.url(
            "restore_point_collection_name", restore_point_collection_name, "str"
        ),
        "vmRestorePointName": _SERIALIZER.url("vm_restore_point_name", vm_restore_point_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_disk_restore_point_grant_access_request(  # pylint: disable=name-too-long
    resource_group_name: str,
    restore_point_collection_name: str,
    vm_restore_point_name: str,
    disk_restore_point_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/restorePointCollections/{restorePointCollectionName}/restorePoints/{vmRestorePointName}/diskRestorePoints/{diskRestorePointName}/beginGetAccess",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "restorePointCollectionName": _SERIALIZER.url(
            "restore_point_collection_name", restore_point_collection_name, "str"
        ),
        "vmRestorePointName": _SERIALIZER.url("vm_restore_point_name", vm_restore_point_name, "str"),
        "diskRestorePointName": _SERIALIZER.url("disk_restore_point_name", disk_restore_point_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="POST", url=_url, params=_params, headers=_headers, **kwargs)


def build_disk_restore_point_revoke_access_request(  # pylint: disable=name-too-long
    resource_group_name: str,
    restore_point_collection_name: str,
    vm_restore_point_name: str,
    disk_restore_point_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/restorePointCollections/{restorePointCollectionName}/restorePoints/{vmRestorePointName}/diskRestorePoints/{diskRestorePointName}/endGetAccess",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "restorePointCollectionName": _SERIALIZER.url(
            "restore_point_collection_name", restore_point_collection_name, "str"
        ),
        "vmRestorePointName": _SERIALIZER.url("vm_restore_point_name", vm_restore_point_name, "str"),
        "diskRestorePointName": _SERIALIZER.url("disk_restore_point_name", disk_restore_point_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="POST", url=_url, params=_params, headers=_headers, **kwargs)


def build_snapshots_create_or_update_request(
    resource_group_name: str, snapshot_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/snapshots/{snapshotName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "snapshotName": _SERIALIZER.url("snapshot_name", snapshot_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_snapshots_update_request(
    resource_group_name: str, snapshot_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/snapshots/{snapshotName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "snapshotName": _SERIALIZER.url("snapshot_name", snapshot_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_snapshots_get_request(
    resource_group_name: str, snapshot_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/snapshots/{snapshotName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "snapshotName": _SERIALIZER.url("snapshot_name", snapshot_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_snapshots_delete_request(
    resource_group_name: str, snapshot_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})

    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-03-02"))
    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/snapshots/{snapshotName}",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "snapshotName": _SERIALIZER.url("snapshot_name", snapshot_name, "str"),
    }

    _url: str = _url.format(**path_format_arguments)  # type: ignore

    # Construct parameters
    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")

    return HttpRequest(method="DELETE", url=_url, params=_params, **kwargs)


def build_snapshots_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", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/snapshots",
    )  # pylint: disable=line-too-long
    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_snapshots_list_request(subscription_id: str, **kwargs: Any) -> HttpRequest:
    _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})

    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-03-02"))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop("template_url", "/subscriptions/{subscriptionId}/providers/Microsoft.Compute/snapshots")
    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_snapshots_grant_access_request(
    resource_group_name: str, snapshot_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", "2024-03-02"))
    content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
    accept = _headers.pop("Accept", "application/json")

    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/snapshots/{snapshotName}/beginGetAccess",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "snapshotName": _SERIALIZER.url("snapshot_name", snapshot_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="POST", url=_url, params=_params, headers=_headers, **kwargs)


def build_snapshots_revoke_access_request(
    resource_group_name: str, snapshot_name: str, subscription_id: str, **kwargs: Any
) -> HttpRequest:
    _params = case_insensitive_dict(kwargs.pop("params", {}) or {})

    api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2024-03-02"))
    # Construct URL
    _url = kwargs.pop(
        "template_url",
        "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.Compute/snapshots/{snapshotName}/endGetAccess",
    )  # pylint: disable=line-too-long
    path_format_arguments = {
        "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"),
        "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"),
        "snapshotName": _SERIALIZER.url("snapshot_name", snapshot_name, "str"),
    }

    _url: str = _url.format(**path_format_arguments)  # type: ignore

    # Construct parameters
    _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str")

    return HttpRequest(method="POST", url=_url, params=_params, **kwargs)


[docs] class DisksOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.compute.v2024_03_02.ComputeManagementClient`'s :attr:`disks` attribute. """ models = _models def __init__(self, *args, **kwargs): input_args = list(args) self._client = input_args.pop(0) if input_args else kwargs.pop("client") self._config = input_args.pop(0) if input_args else kwargs.pop("config") self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer") self._api_version = input_args.pop(0) if input_args else kwargs.pop("api_version") def _create_or_update_initial( self, resource_group_name: str, disk_name: str, disk: Union[_models.Disk, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(disk, (IOBase, bytes)): _content = disk else: _json = self._serialize.body(disk, "Disk") _request = build_disks_create_or_update_request( resource_group_name=resource_group_name, disk_name=disk_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_create_or_update( self, resource_group_name: str, disk_name: str, disk: _models.Disk, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.Disk]: """Creates or updates a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param disk: Disk object supplied in the body of the Put disk operation. Required. :type disk: ~azure.mgmt.compute.v2024_03_02.models.Disk :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 either Disk or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Disk] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, disk_name: str, disk: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.Disk]: """Creates or updates a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param disk: Disk object supplied in the body of the Put disk operation. Required. :type disk: 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 either Disk or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Disk] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_create_or_update( self, resource_group_name: str, disk_name: str, disk: Union[_models.Disk, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.Disk]: """Creates or updates a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param disk: Disk object supplied in the body of the Put disk operation. Is either a Disk type or a IO[bytes] type. Required. :type disk: ~azure.mgmt.compute.v2024_03_02.models.Disk or IO[bytes] :return: An instance of LROPoller that returns either Disk or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Disk] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.Disk] = 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, disk_name=disk_name, disk=disk, api_version=api_version, 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): deserialized = self._deserialize("Disk", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.Disk].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.Disk]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
def _update_initial( self, resource_group_name: str, disk_name: str, disk: Union[_models.DiskUpdate, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(disk, (IOBase, bytes)): _content = disk else: _json = self._serialize.body(disk, "DiskUpdate") _request = build_disks_update_request( resource_group_name=resource_group_name, disk_name=disk_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_update( self, resource_group_name: str, disk_name: str, disk: _models.DiskUpdate, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.Disk]: """Updates (patches) a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param disk: Disk object supplied in the body of the Patch disk operation. Required. :type disk: ~azure.mgmt.compute.v2024_03_02.models.DiskUpdate :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 either Disk or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Disk] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, disk_name: str, disk: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.Disk]: """Updates (patches) a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param disk: Disk object supplied in the body of the Patch disk operation. Required. :type disk: 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 either Disk or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Disk] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_update( self, resource_group_name: str, disk_name: str, disk: Union[_models.DiskUpdate, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.Disk]: """Updates (patches) a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param disk: Disk object supplied in the body of the Patch disk operation. Is either a DiskUpdate type or a IO[bytes] type. Required. :type disk: ~azure.mgmt.compute.v2024_03_02.models.DiskUpdate or IO[bytes] :return: An instance of LROPoller that returns either Disk or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Disk] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.Disk] = 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._update_initial( resource_group_name=resource_group_name, disk_name=disk_name, disk=disk, api_version=api_version, 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): deserialized = self._deserialize("Disk", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.Disk].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.Disk]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
[docs] @distributed_trace def get(self, resource_group_name: str, disk_name: str, **kwargs: Any) -> _models.Disk: """Gets information about a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :return: Disk or the result of cls(response) :rtype: ~azure.mgmt.compute.v2024_03_02.models.Disk :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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.Disk] = kwargs.pop("cls", None) _request = build_disks_get_request( resource_group_name=resource_group_name, disk_name=disk_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = self._deserialize("Disk", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
def _delete_initial(self, resource_group_name: str, disk_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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_disks_delete_request( resource_group_name=resource_group_name, disk_name=disk_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def begin_delete(self, resource_group_name: str, disk_name: str, **kwargs: Any) -> LROPoller[None]: """Deletes a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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, disk_name=disk_name, api_version=api_version, 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 if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **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
[docs] @distributed_trace def list_by_resource_group(self, resource_group_name: str, **kwargs: Any) -> Iterable["_models.Disk"]: """Lists all the disks under a resource group. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :return: An iterator like instance of either Disk or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.Disk] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskList] = 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_disks_list_by_resource_group_request( resource_group_name=resource_group_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("DiskList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
[docs] @distributed_trace def list(self, **kwargs: Any) -> Iterable["_models.Disk"]: """Lists all the disks under a subscription. :return: An iterator like instance of either Disk or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.Disk] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskList] = 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_disks_list_request( subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("DiskList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
def _grant_access_initial( self, resource_group_name: str, disk_name: str, grant_access_data: Union[_models.GrantAccessData, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(grant_access_data, (IOBase, bytes)): _content = grant_access_data else: _json = self._serialize.body(grant_access_data, "GrantAccessData") _request = build_disks_grant_access_request( resource_group_name=resource_group_name, disk_name=disk_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_grant_access( self, resource_group_name: str, disk_name: str, grant_access_data: _models.GrantAccessData, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param grant_access_data: Access data object supplied in the body of the get disk access operation. Required. :type grant_access_data: ~azure.mgmt.compute.v2024_03_02.models.GrantAccessData :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 either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_grant_access( self, resource_group_name: str, disk_name: str, grant_access_data: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param grant_access_data: Access data object supplied in the body of the get disk access operation. Required. :type grant_access_data: 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 either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_grant_access( self, resource_group_name: str, disk_name: str, grant_access_data: Union[_models.GrantAccessData, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :param grant_access_data: Access data object supplied in the body of the get disk access operation. Is either a GrantAccessData type or a IO[bytes] type. Required. :type grant_access_data: ~azure.mgmt.compute.v2024_03_02.models.GrantAccessData or IO[bytes] :return: An instance of LROPoller that returns either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.AccessUri] = 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._grant_access_initial( resource_group_name=resource_group_name, disk_name=disk_name, grant_access_data=grant_access_data, api_version=api_version, 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): deserialized = self._deserialize("AccessUri", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, lro_options={"final-state-via": "location"}, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.AccessUri].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AccessUri]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
def _revoke_access_initial(self, resource_group_name: str, disk_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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_disks_revoke_access_request( resource_group_name=resource_group_name, disk_name=disk_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def begin_revoke_access(self, resource_group_name: str, disk_name: str, **kwargs: Any) -> LROPoller[None]: """Revokes access to a disk. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_name: The name of the managed disk that is being created. The name can't be changed after the disk is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_name: str :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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._revoke_access_initial( resource_group_name=resource_group_name, disk_name=disk_name, api_version=api_version, 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 if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, lro_options={"final-state-via": "location"}, **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
[docs] class DiskAccessesOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.compute.v2024_03_02.ComputeManagementClient`'s :attr:`disk_accesses` attribute. """ models = _models def __init__(self, *args, **kwargs): input_args = list(args) self._client = input_args.pop(0) if input_args else kwargs.pop("client") self._config = input_args.pop(0) if input_args else kwargs.pop("config") self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer") self._api_version = input_args.pop(0) if input_args else kwargs.pop("api_version") def _create_or_update_initial( self, resource_group_name: str, disk_access_name: str, disk_access: Union[_models.DiskAccess, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(disk_access, (IOBase, bytes)): _content = disk_access else: _json = self._serialize.body(disk_access, "DiskAccess") _request = build_disk_accesses_create_or_update_request( resource_group_name=resource_group_name, disk_access_name=disk_access_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_create_or_update( self, resource_group_name: str, disk_access_name: str, disk_access: _models.DiskAccess, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DiskAccess]: """Creates or updates a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param disk_access: disk access object supplied in the body of the Put disk access operation. Required. :type disk_access: ~azure.mgmt.compute.v2024_03_02.models.DiskAccess :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 either DiskAccess or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskAccess] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, disk_access_name: str, disk_access: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DiskAccess]: """Creates or updates a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param disk_access: disk access object supplied in the body of the Put disk access operation. Required. :type disk_access: 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 either DiskAccess or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskAccess] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_create_or_update( self, resource_group_name: str, disk_access_name: str, disk_access: Union[_models.DiskAccess, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DiskAccess]: """Creates or updates a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param disk_access: disk access object supplied in the body of the Put disk access operation. Is either a DiskAccess type or a IO[bytes] type. Required. :type disk_access: ~azure.mgmt.compute.v2024_03_02.models.DiskAccess or IO[bytes] :return: An instance of LROPoller that returns either DiskAccess or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskAccess] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.DiskAccess] = 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, disk_access_name=disk_access_name, disk_access=disk_access, api_version=api_version, 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): deserialized = self._deserialize("DiskAccess", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.DiskAccess].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DiskAccess]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
def _update_initial( self, resource_group_name: str, disk_access_name: str, disk_access: Union[_models.DiskAccessUpdate, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(disk_access, (IOBase, bytes)): _content = disk_access else: _json = self._serialize.body(disk_access, "DiskAccessUpdate") _request = build_disk_accesses_update_request( resource_group_name=resource_group_name, disk_access_name=disk_access_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_update( self, resource_group_name: str, disk_access_name: str, disk_access: _models.DiskAccessUpdate, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DiskAccess]: """Updates (patches) a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param disk_access: disk access object supplied in the body of the Patch disk access operation. Required. :type disk_access: ~azure.mgmt.compute.v2024_03_02.models.DiskAccessUpdate :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 either DiskAccess or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskAccess] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, disk_access_name: str, disk_access: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DiskAccess]: """Updates (patches) a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param disk_access: disk access object supplied in the body of the Patch disk access operation. Required. :type disk_access: 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 either DiskAccess or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskAccess] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_update( self, resource_group_name: str, disk_access_name: str, disk_access: Union[_models.DiskAccessUpdate, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DiskAccess]: """Updates (patches) a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param disk_access: disk access object supplied in the body of the Patch disk access operation. Is either a DiskAccessUpdate type or a IO[bytes] type. Required. :type disk_access: ~azure.mgmt.compute.v2024_03_02.models.DiskAccessUpdate or IO[bytes] :return: An instance of LROPoller that returns either DiskAccess or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskAccess] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.DiskAccess] = 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._update_initial( resource_group_name=resource_group_name, disk_access_name=disk_access_name, disk_access=disk_access, api_version=api_version, 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): deserialized = self._deserialize("DiskAccess", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.DiskAccess].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DiskAccess]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
[docs] @distributed_trace def get(self, resource_group_name: str, disk_access_name: str, **kwargs: Any) -> _models.DiskAccess: """Gets information about a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :return: DiskAccess or the result of cls(response) :rtype: ~azure.mgmt.compute.v2024_03_02.models.DiskAccess :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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskAccess] = kwargs.pop("cls", None) _request = build_disk_accesses_get_request( resource_group_name=resource_group_name, disk_access_name=disk_access_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = self._deserialize("DiskAccess", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
def _delete_initial(self, resource_group_name: str, disk_access_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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_disk_accesses_delete_request( resource_group_name=resource_group_name, disk_access_name=disk_access_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def begin_delete(self, resource_group_name: str, disk_access_name: str, **kwargs: Any) -> LROPoller[None]: """Deletes a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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, disk_access_name=disk_access_name, api_version=api_version, 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 if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **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
[docs] @distributed_trace def list_by_resource_group(self, resource_group_name: str, **kwargs: Any) -> Iterable["_models.DiskAccess"]: """Lists all the disk access resources under a resource group. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :return: An iterator like instance of either DiskAccess or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.DiskAccess] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskAccessList] = 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_disk_accesses_list_by_resource_group_request( resource_group_name=resource_group_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("DiskAccessList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
[docs] @distributed_trace def list(self, **kwargs: Any) -> Iterable["_models.DiskAccess"]: """Lists all the disk access resources under a subscription. :return: An iterator like instance of either DiskAccess or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.DiskAccess] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskAccessList] = 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_disk_accesses_list_request( subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("DiskAccessList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
def _update_a_private_endpoint_connection_initial( # pylint: disable=name-too-long self, resource_group_name: str, disk_access_name: str, private_endpoint_connection_name: str, private_endpoint_connection: Union[_models.PrivateEndpointConnection, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(private_endpoint_connection, (IOBase, bytes)): _content = private_endpoint_connection else: _json = self._serialize.body(private_endpoint_connection, "PrivateEndpointConnection") _request = build_disk_accesses_update_a_private_endpoint_connection_request( resource_group_name=resource_group_name, disk_access_name=disk_access_name, private_endpoint_connection_name=private_endpoint_connection_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_update_a_private_endpoint_connection( # pylint: disable=name-too-long self, resource_group_name: str, disk_access_name: str, private_endpoint_connection_name: str, private_endpoint_connection: _models.PrivateEndpointConnection, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.PrivateEndpointConnection]: """Approve or reject a private endpoint connection under disk access resource, this can't be used to create a new private endpoint connection. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param private_endpoint_connection_name: The name of the private endpoint connection. Required. :type private_endpoint_connection_name: str :param private_endpoint_connection: private endpoint connection object supplied in the body of the Put private endpoint connection operation. Required. :type private_endpoint_connection: ~azure.mgmt.compute.v2024_03_02.models.PrivateEndpointConnection :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 either PrivateEndpointConnection or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.PrivateEndpointConnection] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update_a_private_endpoint_connection( # pylint: disable=name-too-long self, resource_group_name: str, disk_access_name: str, private_endpoint_connection_name: str, private_endpoint_connection: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.PrivateEndpointConnection]: """Approve or reject a private endpoint connection under disk access resource, this can't be used to create a new private endpoint connection. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param private_endpoint_connection_name: The name of the private endpoint connection. Required. :type private_endpoint_connection_name: str :param private_endpoint_connection: private endpoint connection object supplied in the body of the Put private endpoint connection operation. Required. :type private_endpoint_connection: 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 either PrivateEndpointConnection or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.PrivateEndpointConnection] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_update_a_private_endpoint_connection( # pylint: disable=name-too-long self, resource_group_name: str, disk_access_name: str, private_endpoint_connection_name: str, private_endpoint_connection: Union[_models.PrivateEndpointConnection, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.PrivateEndpointConnection]: """Approve or reject a private endpoint connection under disk access resource, this can't be used to create a new private endpoint connection. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param private_endpoint_connection_name: The name of the private endpoint connection. Required. :type private_endpoint_connection_name: str :param private_endpoint_connection: private endpoint connection object supplied in the body of the Put private endpoint connection operation. Is either a PrivateEndpointConnection type or a IO[bytes] type. Required. :type private_endpoint_connection: ~azure.mgmt.compute.v2024_03_02.models.PrivateEndpointConnection or IO[bytes] :return: An instance of LROPoller that returns either PrivateEndpointConnection or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.PrivateEndpointConnection] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.PrivateEndpointConnection] = 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._update_a_private_endpoint_connection_initial( resource_group_name=resource_group_name, disk_access_name=disk_access_name, private_endpoint_connection_name=private_endpoint_connection_name, private_endpoint_connection=private_endpoint_connection, api_version=api_version, 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): deserialized = self._deserialize("PrivateEndpointConnection", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.PrivateEndpointConnection].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.PrivateEndpointConnection]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
[docs] @distributed_trace def get_a_private_endpoint_connection( self, resource_group_name: str, disk_access_name: str, private_endpoint_connection_name: str, **kwargs: Any ) -> _models.PrivateEndpointConnection: """Gets information about a private endpoint connection under a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param private_endpoint_connection_name: The name of the private endpoint connection. Required. :type private_endpoint_connection_name: str :return: PrivateEndpointConnection or the result of cls(response) :rtype: ~azure.mgmt.compute.v2024_03_02.models.PrivateEndpointConnection :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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.PrivateEndpointConnection] = kwargs.pop("cls", None) _request = build_disk_accesses_get_a_private_endpoint_connection_request( resource_group_name=resource_group_name, disk_access_name=disk_access_name, private_endpoint_connection_name=private_endpoint_connection_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = self._deserialize("PrivateEndpointConnection", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
def _delete_a_private_endpoint_connection_initial( # pylint: disable=name-too-long self, resource_group_name: str, disk_access_name: str, private_endpoint_connection_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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_disk_accesses_delete_a_private_endpoint_connection_request( resource_group_name=resource_group_name, disk_access_name=disk_access_name, private_endpoint_connection_name=private_endpoint_connection_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def begin_delete_a_private_endpoint_connection( # pylint: disable=name-too-long self, resource_group_name: str, disk_access_name: str, private_endpoint_connection_name: str, **kwargs: Any ) -> LROPoller[None]: """Deletes a private endpoint connection under a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :param private_endpoint_connection_name: The name of the private endpoint connection. Required. :type private_endpoint_connection_name: str :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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_a_private_endpoint_connection_initial( resource_group_name=resource_group_name, disk_access_name=disk_access_name, private_endpoint_connection_name=private_endpoint_connection_name, api_version=api_version, 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 if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **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
[docs] @distributed_trace def list_private_endpoint_connections( self, resource_group_name: str, disk_access_name: str, **kwargs: Any ) -> Iterable["_models.PrivateEndpointConnection"]: """List information about private endpoint connections under a disk access resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_access_name: The name of the disk access resource that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_access_name: str :return: An iterator like instance of either PrivateEndpointConnection or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.PrivateEndpointConnection] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.PrivateEndpointConnectionListResult] = 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_disk_accesses_list_private_endpoint_connections_request( resource_group_name=resource_group_name, disk_access_name=disk_access_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("PrivateEndpointConnectionListResult", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
[docs] class DiskEncryptionSetsOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.compute.v2024_03_02.ComputeManagementClient`'s :attr:`disk_encryption_sets` attribute. """ models = _models def __init__(self, *args, **kwargs): input_args = list(args) self._client = input_args.pop(0) if input_args else kwargs.pop("client") self._config = input_args.pop(0) if input_args else kwargs.pop("config") self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer") self._api_version = input_args.pop(0) if input_args else kwargs.pop("api_version") def _create_or_update_initial( self, resource_group_name: str, disk_encryption_set_name: str, disk_encryption_set: Union[_models.DiskEncryptionSet, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(disk_encryption_set, (IOBase, bytes)): _content = disk_encryption_set else: _json = self._serialize.body(disk_encryption_set, "DiskEncryptionSet") _request = build_disk_encryption_sets_create_or_update_request( resource_group_name=resource_group_name, disk_encryption_set_name=disk_encryption_set_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_create_or_update( self, resource_group_name: str, disk_encryption_set_name: str, disk_encryption_set: _models.DiskEncryptionSet, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DiskEncryptionSet]: """Creates or updates a disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :param disk_encryption_set: disk encryption set object supplied in the body of the Put disk encryption set operation. Required. :type disk_encryption_set: ~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet :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 either DiskEncryptionSet or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, disk_encryption_set_name: str, disk_encryption_set: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DiskEncryptionSet]: """Creates or updates a disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :param disk_encryption_set: disk encryption set object supplied in the body of the Put disk encryption set operation. Required. :type disk_encryption_set: 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 either DiskEncryptionSet or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_create_or_update( self, resource_group_name: str, disk_encryption_set_name: str, disk_encryption_set: Union[_models.DiskEncryptionSet, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DiskEncryptionSet]: """Creates or updates a disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :param disk_encryption_set: disk encryption set object supplied in the body of the Put disk encryption set operation. Is either a DiskEncryptionSet type or a IO[bytes] type. Required. :type disk_encryption_set: ~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet or IO[bytes] :return: An instance of LROPoller that returns either DiskEncryptionSet or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.DiskEncryptionSet] = 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, disk_encryption_set_name=disk_encryption_set_name, disk_encryption_set=disk_encryption_set, api_version=api_version, 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): deserialized = self._deserialize("DiskEncryptionSet", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.DiskEncryptionSet].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DiskEncryptionSet]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
def _update_initial( self, resource_group_name: str, disk_encryption_set_name: str, disk_encryption_set: Union[_models.DiskEncryptionSetUpdate, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(disk_encryption_set, (IOBase, bytes)): _content = disk_encryption_set else: _json = self._serialize.body(disk_encryption_set, "DiskEncryptionSetUpdate") _request = build_disk_encryption_sets_update_request( resource_group_name=resource_group_name, disk_encryption_set_name=disk_encryption_set_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_update( self, resource_group_name: str, disk_encryption_set_name: str, disk_encryption_set: _models.DiskEncryptionSetUpdate, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DiskEncryptionSet]: """Updates (patches) a disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :param disk_encryption_set: disk encryption set object supplied in the body of the Patch disk encryption set operation. Required. :type disk_encryption_set: ~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSetUpdate :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 either DiskEncryptionSet or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, disk_encryption_set_name: str, disk_encryption_set: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.DiskEncryptionSet]: """Updates (patches) a disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :param disk_encryption_set: disk encryption set object supplied in the body of the Patch disk encryption set operation. Required. :type disk_encryption_set: 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 either DiskEncryptionSet or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_update( self, resource_group_name: str, disk_encryption_set_name: str, disk_encryption_set: Union[_models.DiskEncryptionSetUpdate, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.DiskEncryptionSet]: """Updates (patches) a disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :param disk_encryption_set: disk encryption set object supplied in the body of the Patch disk encryption set operation. Is either a DiskEncryptionSetUpdate type or a IO[bytes] type. Required. :type disk_encryption_set: ~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSetUpdate or IO[bytes] :return: An instance of LROPoller that returns either DiskEncryptionSet or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.DiskEncryptionSet] = 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._update_initial( resource_group_name=resource_group_name, disk_encryption_set_name=disk_encryption_set_name, disk_encryption_set=disk_encryption_set, api_version=api_version, 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): deserialized = self._deserialize("DiskEncryptionSet", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.DiskEncryptionSet].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.DiskEncryptionSet]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
[docs] @distributed_trace def get(self, resource_group_name: str, disk_encryption_set_name: str, **kwargs: Any) -> _models.DiskEncryptionSet: """Gets information about a disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :return: DiskEncryptionSet or the result of cls(response) :rtype: ~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet :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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskEncryptionSet] = kwargs.pop("cls", None) _request = build_disk_encryption_sets_get_request( resource_group_name=resource_group_name, disk_encryption_set_name=disk_encryption_set_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = self._deserialize("DiskEncryptionSet", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
def _delete_initial( self, resource_group_name: str, disk_encryption_set_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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_disk_encryption_sets_delete_request( resource_group_name=resource_group_name, disk_encryption_set_name=disk_encryption_set_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def begin_delete(self, resource_group_name: str, disk_encryption_set_name: str, **kwargs: Any) -> LROPoller[None]: """Deletes a disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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, disk_encryption_set_name=disk_encryption_set_name, api_version=api_version, 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 if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **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
[docs] @distributed_trace def list_by_resource_group(self, resource_group_name: str, **kwargs: Any) -> Iterable["_models.DiskEncryptionSet"]: """Lists all the disk encryption sets under a resource group. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :return: An iterator like instance of either DiskEncryptionSet or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskEncryptionSetList] = 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_disk_encryption_sets_list_by_resource_group_request( resource_group_name=resource_group_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("DiskEncryptionSetList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
[docs] @distributed_trace def list(self, **kwargs: Any) -> Iterable["_models.DiskEncryptionSet"]: """Lists all the disk encryption sets under a subscription. :return: An iterator like instance of either DiskEncryptionSet or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.DiskEncryptionSet] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskEncryptionSetList] = 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_disk_encryption_sets_list_request( subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("DiskEncryptionSetList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
[docs] @distributed_trace def list_associated_resources( self, resource_group_name: str, disk_encryption_set_name: str, **kwargs: Any ) -> Iterable[str]: """Lists all resources that are encrypted with this disk encryption set. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param disk_encryption_set_name: The name of the disk encryption set that is being created. The name can't be changed after the disk encryption set is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The maximum name length is 80 characters. Required. :type disk_encryption_set_name: str :return: An iterator like instance of either str or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[str] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.ResourceUriList] = 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_disk_encryption_sets_list_associated_resources_request( resource_group_name=resource_group_name, disk_encryption_set_name=disk_encryption_set_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("ResourceUriList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
[docs] class DiskRestorePointOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.compute.v2024_03_02.ComputeManagementClient`'s :attr:`disk_restore_point` attribute. """ models = _models def __init__(self, *args, **kwargs): input_args = list(args) self._client = input_args.pop(0) if input_args else kwargs.pop("client") self._config = input_args.pop(0) if input_args else kwargs.pop("config") self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer") self._api_version = input_args.pop(0) if input_args else kwargs.pop("api_version")
[docs] @distributed_trace def get( self, resource_group_name: str, restore_point_collection_name: str, vm_restore_point_name: str, disk_restore_point_name: str, **kwargs: Any ) -> _models.DiskRestorePoint: """Get disk restorePoint resource. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param restore_point_collection_name: The name of the restore point collection that the disk restore point belongs. Required. :type restore_point_collection_name: str :param vm_restore_point_name: The name of the vm restore point that the disk disk restore point belongs. Required. :type vm_restore_point_name: str :param disk_restore_point_name: The name of the disk restore point created. Required. :type disk_restore_point_name: str :return: DiskRestorePoint or the result of cls(response) :rtype: ~azure.mgmt.compute.v2024_03_02.models.DiskRestorePoint :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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskRestorePoint] = kwargs.pop("cls", None) _request = build_disk_restore_point_get_request( resource_group_name=resource_group_name, restore_point_collection_name=restore_point_collection_name, vm_restore_point_name=vm_restore_point_name, disk_restore_point_name=disk_restore_point_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = self._deserialize("DiskRestorePoint", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def list_by_restore_point( self, resource_group_name: str, restore_point_collection_name: str, vm_restore_point_name: str, **kwargs: Any ) -> Iterable["_models.DiskRestorePoint"]: """Lists diskRestorePoints under a vmRestorePoint. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param restore_point_collection_name: The name of the restore point collection that the disk restore point belongs. Required. :type restore_point_collection_name: str :param vm_restore_point_name: The name of the vm restore point that the disk disk restore point belongs. Required. :type vm_restore_point_name: str :return: An iterator like instance of either DiskRestorePoint or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.DiskRestorePoint] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.DiskRestorePointList] = 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_disk_restore_point_list_by_restore_point_request( resource_group_name=resource_group_name, restore_point_collection_name=restore_point_collection_name, vm_restore_point_name=vm_restore_point_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("DiskRestorePointList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
def _grant_access_initial( self, resource_group_name: str, restore_point_collection_name: str, vm_restore_point_name: str, disk_restore_point_name: str, grant_access_data: Union[_models.GrantAccessData, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(grant_access_data, (IOBase, bytes)): _content = grant_access_data else: _json = self._serialize.body(grant_access_data, "GrantAccessData") _request = build_disk_restore_point_grant_access_request( resource_group_name=resource_group_name, restore_point_collection_name=restore_point_collection_name, vm_restore_point_name=vm_restore_point_name, disk_restore_point_name=disk_restore_point_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_grant_access( self, resource_group_name: str, restore_point_collection_name: str, vm_restore_point_name: str, disk_restore_point_name: str, grant_access_data: _models.GrantAccessData, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a diskRestorePoint. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param restore_point_collection_name: The name of the restore point collection that the disk restore point belongs. Required. :type restore_point_collection_name: str :param vm_restore_point_name: The name of the vm restore point that the disk disk restore point belongs. Required. :type vm_restore_point_name: str :param disk_restore_point_name: The name of the disk restore point created. Required. :type disk_restore_point_name: str :param grant_access_data: Access data object supplied in the body of the get disk access operation. Required. :type grant_access_data: ~azure.mgmt.compute.v2024_03_02.models.GrantAccessData :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 either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_grant_access( self, resource_group_name: str, restore_point_collection_name: str, vm_restore_point_name: str, disk_restore_point_name: str, grant_access_data: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a diskRestorePoint. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param restore_point_collection_name: The name of the restore point collection that the disk restore point belongs. Required. :type restore_point_collection_name: str :param vm_restore_point_name: The name of the vm restore point that the disk disk restore point belongs. Required. :type vm_restore_point_name: str :param disk_restore_point_name: The name of the disk restore point created. Required. :type disk_restore_point_name: str :param grant_access_data: Access data object supplied in the body of the get disk access operation. Required. :type grant_access_data: 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 either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_grant_access( self, resource_group_name: str, restore_point_collection_name: str, vm_restore_point_name: str, disk_restore_point_name: str, grant_access_data: Union[_models.GrantAccessData, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a diskRestorePoint. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param restore_point_collection_name: The name of the restore point collection that the disk restore point belongs. Required. :type restore_point_collection_name: str :param vm_restore_point_name: The name of the vm restore point that the disk disk restore point belongs. Required. :type vm_restore_point_name: str :param disk_restore_point_name: The name of the disk restore point created. Required. :type disk_restore_point_name: str :param grant_access_data: Access data object supplied in the body of the get disk access operation. Is either a GrantAccessData type or a IO[bytes] type. Required. :type grant_access_data: ~azure.mgmt.compute.v2024_03_02.models.GrantAccessData or IO[bytes] :return: An instance of LROPoller that returns either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.AccessUri] = 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._grant_access_initial( resource_group_name=resource_group_name, restore_point_collection_name=restore_point_collection_name, vm_restore_point_name=vm_restore_point_name, disk_restore_point_name=disk_restore_point_name, grant_access_data=grant_access_data, api_version=api_version, 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): deserialized = self._deserialize("AccessUri", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, lro_options={"final-state-via": "location"}, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.AccessUri].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AccessUri]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
def _revoke_access_initial( self, resource_group_name: str, restore_point_collection_name: str, vm_restore_point_name: str, disk_restore_point_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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_disk_restore_point_revoke_access_request( resource_group_name=resource_group_name, restore_point_collection_name=restore_point_collection_name, vm_restore_point_name=vm_restore_point_name, disk_restore_point_name=disk_restore_point_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def begin_revoke_access( self, resource_group_name: str, restore_point_collection_name: str, vm_restore_point_name: str, disk_restore_point_name: str, **kwargs: Any ) -> LROPoller[None]: """Revokes access to a diskRestorePoint. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param restore_point_collection_name: The name of the restore point collection that the disk restore point belongs. Required. :type restore_point_collection_name: str :param vm_restore_point_name: The name of the vm restore point that the disk disk restore point belongs. Required. :type vm_restore_point_name: str :param disk_restore_point_name: The name of the disk restore point created. Required. :type disk_restore_point_name: str :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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._revoke_access_initial( resource_group_name=resource_group_name, restore_point_collection_name=restore_point_collection_name, vm_restore_point_name=vm_restore_point_name, disk_restore_point_name=disk_restore_point_name, api_version=api_version, 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 if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, lro_options={"final-state-via": "location"}, **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
[docs] class SnapshotsOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.compute.v2024_03_02.ComputeManagementClient`'s :attr:`snapshots` attribute. """ models = _models def __init__(self, *args, **kwargs): input_args = list(args) self._client = input_args.pop(0) if input_args else kwargs.pop("client") self._config = input_args.pop(0) if input_args else kwargs.pop("config") self._serialize = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize = input_args.pop(0) if input_args else kwargs.pop("deserializer") self._api_version = input_args.pop(0) if input_args else kwargs.pop("api_version") def _create_or_update_initial( self, resource_group_name: str, snapshot_name: str, snapshot: Union[_models.Snapshot, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(snapshot, (IOBase, bytes)): _content = snapshot else: _json = self._serialize.body(snapshot, "Snapshot") _request = build_snapshots_create_or_update_request( resource_group_name=resource_group_name, snapshot_name=snapshot_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_create_or_update( self, resource_group_name: str, snapshot_name: str, snapshot: _models.Snapshot, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.Snapshot]: """Creates or updates a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param snapshot: Snapshot object supplied in the body of the Put disk operation. Required. :type snapshot: ~azure.mgmt.compute.v2024_03_02.models.Snapshot :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 either Snapshot or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Snapshot] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, snapshot_name: str, snapshot: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.Snapshot]: """Creates or updates a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param snapshot: Snapshot object supplied in the body of the Put disk operation. Required. :type snapshot: 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 either Snapshot or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Snapshot] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_create_or_update( self, resource_group_name: str, snapshot_name: str, snapshot: Union[_models.Snapshot, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.Snapshot]: """Creates or updates a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param snapshot: Snapshot object supplied in the body of the Put disk operation. Is either a Snapshot type or a IO[bytes] type. Required. :type snapshot: ~azure.mgmt.compute.v2024_03_02.models.Snapshot or IO[bytes] :return: An instance of LROPoller that returns either Snapshot or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Snapshot] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.Snapshot] = 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, snapshot_name=snapshot_name, snapshot=snapshot, api_version=api_version, 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): deserialized = self._deserialize("Snapshot", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.Snapshot].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.Snapshot]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
def _update_initial( self, resource_group_name: str, snapshot_name: str, snapshot: Union[_models.SnapshotUpdate, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(snapshot, (IOBase, bytes)): _content = snapshot else: _json = self._serialize.body(snapshot, "SnapshotUpdate") _request = build_snapshots_update_request( resource_group_name=resource_group_name, snapshot_name=snapshot_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_update( self, resource_group_name: str, snapshot_name: str, snapshot: _models.SnapshotUpdate, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.Snapshot]: """Updates (patches) a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param snapshot: Snapshot object supplied in the body of the Patch snapshot operation. Required. :type snapshot: ~azure.mgmt.compute.v2024_03_02.models.SnapshotUpdate :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 either Snapshot or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Snapshot] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, snapshot_name: str, snapshot: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.Snapshot]: """Updates (patches) a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param snapshot: Snapshot object supplied in the body of the Patch snapshot operation. Required. :type snapshot: 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 either Snapshot or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Snapshot] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_update( self, resource_group_name: str, snapshot_name: str, snapshot: Union[_models.SnapshotUpdate, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.Snapshot]: """Updates (patches) a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param snapshot: Snapshot object supplied in the body of the Patch snapshot operation. Is either a SnapshotUpdate type or a IO[bytes] type. Required. :type snapshot: ~azure.mgmt.compute.v2024_03_02.models.SnapshotUpdate or IO[bytes] :return: An instance of LROPoller that returns either Snapshot or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.Snapshot] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.Snapshot] = 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._update_initial( resource_group_name=resource_group_name, snapshot_name=snapshot_name, snapshot=snapshot, api_version=api_version, 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): deserialized = self._deserialize("Snapshot", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **kwargs)) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.Snapshot].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.Snapshot]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
[docs] @distributed_trace def get(self, resource_group_name: str, snapshot_name: str, **kwargs: Any) -> _models.Snapshot: """Gets information about a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :return: Snapshot or the result of cls(response) :rtype: ~azure.mgmt.compute.v2024_03_02.models.Snapshot :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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.Snapshot] = kwargs.pop("cls", None) _request = build_snapshots_get_request( resource_group_name=resource_group_name, snapshot_name=snapshot_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = self._deserialize("Snapshot", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
def _delete_initial(self, resource_group_name: str, snapshot_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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_snapshots_delete_request( resource_group_name=resource_group_name, snapshot_name=snapshot_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def begin_delete(self, resource_group_name: str, snapshot_name: str, **kwargs: Any) -> LROPoller[None]: """Deletes a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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, snapshot_name=snapshot_name, api_version=api_version, 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 if polling is True: polling_method: PollingMethod = cast(PollingMethod, ARMPolling(lro_delay, **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
[docs] @distributed_trace def list_by_resource_group(self, resource_group_name: str, **kwargs: Any) -> Iterable["_models.Snapshot"]: """Lists snapshots under a resource group. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :return: An iterator like instance of either Snapshot or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.Snapshot] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.SnapshotList] = 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_snapshots_list_by_resource_group_request( resource_group_name=resource_group_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("SnapshotList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
[docs] @distributed_trace def list(self, **kwargs: Any) -> Iterable["_models.Snapshot"]: """Lists snapshots under a subscription. :return: An iterator like instance of either Snapshot or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.compute.v2024_03_02.models.Snapshot] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[_models.SnapshotList] = 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_snapshots_list_request( subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) 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._api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) _request.url = self._client.format_url(_request.url) _request.method = "GET" return _request def extract_data(pipeline_response): deserialized = self._deserialize("SnapshotList", pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.next_link 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data)
def _grant_access_initial( self, resource_group_name: str, snapshot_name: str, grant_access_data: Union[_models.GrantAccessData, 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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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" _json = None _content = None if isinstance(grant_access_data, (IOBase, bytes)): _content = grant_access_data else: _json = self._serialize.body(grant_access_data, "GrantAccessData") _request = build_snapshots_grant_access_request( resource_group_name=resource_group_name, snapshot_name=snapshot_name, subscription_id=self._config.subscription_id, api_version=api_version, content_type=content_type, json=_json, content=_content, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @overload def begin_grant_access( self, resource_group_name: str, snapshot_name: str, grant_access_data: _models.GrantAccessData, *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param grant_access_data: Access data object supplied in the body of the get snapshot access operation. Required. :type grant_access_data: ~azure.mgmt.compute.v2024_03_02.models.GrantAccessData :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 either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_grant_access( self, resource_group_name: str, snapshot_name: str, grant_access_data: IO[bytes], *, content_type: str = "application/json", **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param grant_access_data: Access data object supplied in the body of the get snapshot access operation. Required. :type grant_access_data: 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 either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """
[docs] @distributed_trace def begin_grant_access( self, resource_group_name: str, snapshot_name: str, grant_access_data: Union[_models.GrantAccessData, IO[bytes]], **kwargs: Any ) -> LROPoller[_models.AccessUri]: """Grants access to a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :param grant_access_data: Access data object supplied in the body of the get snapshot access operation. Is either a GrantAccessData type or a IO[bytes] type. Required. :type grant_access_data: ~azure.mgmt.compute.v2024_03_02.models.GrantAccessData or IO[bytes] :return: An instance of LROPoller that returns either AccessUri or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.compute.v2024_03_02.models.AccessUri] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.AccessUri] = 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._grant_access_initial( resource_group_name=resource_group_name, snapshot_name=snapshot_name, grant_access_data=grant_access_data, api_version=api_version, 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): deserialized = self._deserialize("AccessUri", pipeline_response.http_response) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, lro_options={"final-state-via": "location"}, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.AccessUri].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AccessUri]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )
def _revoke_access_initial(self, resource_group_name: str, snapshot_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 = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_snapshots_revoke_access_request( resource_group_name=resource_group_name, snapshot_name=snapshot_name, subscription_id=self._config.subscription_id, api_version=api_version, headers=_headers, params=_params, ) _request.url = self._client.format_url(_request.url) _decompress = kwargs.pop("decompress", True) _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, 202]: 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) raise HttpResponseError(response=response, error_format=ARMErrorFormat) deserialized = response.stream_download(self._client._pipeline, decompress=_decompress) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore
[docs] @distributed_trace def begin_revoke_access(self, resource_group_name: str, snapshot_name: str, **kwargs: Any) -> LROPoller[None]: """Revokes access to a snapshot. :param resource_group_name: The name of the resource group. Required. :type resource_group_name: str :param snapshot_name: The name of the snapshot that is being created. The name can't be changed after the snapshot is created. Supported characters for the name are a-z, A-Z, 0-9, _ and -. The max name length is 80 characters. Required. :type snapshot_name: str :return: An instance of LROPoller that returns either None or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", self._api_version or "2024-03-02")) 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._revoke_access_initial( resource_group_name=resource_group_name, snapshot_name=snapshot_name, api_version=api_version, 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 if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, lro_options={"final-state-via": "location"}, **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