# pylint: disable=line-too-long,useless-suppression,too-many-lines # coding=utf-8 # -------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for license information. # Code generated by Microsoft (R) Python Code Generator. # Changes may cause incorrect behavior and will be lost if the code is regenerated. # -------------------------------------------------------------------------- from collections.abc import MutableMapping from io import IOBase import json from typing import Any, Callable, Dict, IO, Iterator, List, Optional, TypeVar, Union, cast, overload import urllib.parse from azure.core import MatchConditions, PipelineClient from azure.core.exceptions import ( ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceModifiedError, ResourceNotFoundError, ResourceNotModifiedError, StreamClosedError, StreamConsumedError, map_error, ) from azure.core.paging import ItemPaged from azure.core.pipeline import PipelineResponse from azure.core.polling import LROPoller, NoPolling, PollingMethod from azure.core.rest import HttpRequest, HttpResponse from azure.core.tracing.decorator import distributed_trace from azure.core.utils import case_insensitive_dict from azure.mgmt.core.exceptions import ARMErrorFormat from azure.mgmt.core.polling.arm_polling import ARMPolling from .. import models as _models from .._configuration import ContainerServiceFleetMgmtClientConfiguration from .._utils.model_base import SdkJSONEncoder, _deserialize, _failsafe_deserialize from .._utils.serialization import Deserializer, Serializer from .._utils.utils import prep_if_match, prep_if_none_match from .._validation import api_version_validation T = TypeVar("T") ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]] JSON = MutableMapping[str, Any] _SERIALIZER = Serializer() _SERIALIZER.client_side_validation = False def build_operations_list_request(**kwargs: Any) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/providers/Microsoft.ContainerService/operations" # Construct parameters _params["api-version"] = _SERIALIZER.query("api_version", api_version, "str") # Construct headers _headers["Accept"] = _SERIALIZER.header("accept", accept, "str") return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs) def build_fleets_get_request( resource_group_name: str, fleet_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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_fleets_begin_create_or_update_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs) def build_fleets_begin_update_request( resource_group_name: str, fleet_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="PATCH", url=_url, params=_params, headers=_headers, **kwargs) def build_fleets_delete_request( resource_group_name: str, fleet_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs) def build_fleets_list_by_resource_group_request( # pylint: disable=name-too-long resource_group_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = ( "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets" ) 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_fleets_list_by_subscription_request( # pylint: disable=name-too-long subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/providers/Microsoft.ContainerService/fleets" 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_fleets_list_credentials_request( resource_group_name: str, fleet_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/listCredentials" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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_fleet_members_get_request( resource_group_name: str, fleet_name: str, fleet_member_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/members/{fleetMemberName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "fleetMemberName": _SERIALIZER.url("fleet_member_name", fleet_member_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_fleet_members_create_request( resource_group_name: str, fleet_name: str, fleet_member_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/members/{fleetMemberName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "fleetMemberName": _SERIALIZER.url("fleet_member_name", fleet_member_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs) def build_fleet_members_begin_update_request( resource_group_name: str, fleet_name: str, fleet_member_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/members/{fleetMemberName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "fleetMemberName": _SERIALIZER.url("fleet_member_name", fleet_member_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="PATCH", url=_url, params=_params, headers=_headers, **kwargs) def build_fleet_members_delete_request( resource_group_name: str, fleet_name: str, fleet_member_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/members/{fleetMemberName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "fleetMemberName": _SERIALIZER.url("fleet_member_name", fleet_member_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs) def build_fleet_members_list_by_fleet_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/members" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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_gates_get_request( resource_group_name: str, fleet_name: str, gate_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/gates/{gateName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "gateName": _SERIALIZER.url("gate_name", gate_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_gates_update_request( resource_group_name: str, fleet_name: str, gate_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/gates/{gateName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "gateName": _SERIALIZER.url("gate_name", gate_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="PATCH", url=_url, params=_params, headers=_headers, **kwargs) def build_gates_list_by_fleet_request( resource_group_name: str, fleet_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/gates" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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_update_runs_get_request( resource_group_name: str, fleet_name: str, update_run_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateRuns/{updateRunName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateRunName": _SERIALIZER.url("update_run_name", update_run_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_update_runs_create_or_update_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, update_run_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateRuns/{updateRunName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateRunName": _SERIALIZER.url("update_run_name", update_run_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs) def build_update_runs_delete_request( resource_group_name: str, fleet_name: str, update_run_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateRuns/{updateRunName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateRunName": _SERIALIZER.url("update_run_name", update_run_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs) def build_update_runs_list_by_fleet_request( resource_group_name: str, fleet_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateRuns" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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_update_runs_start_request( resource_group_name: str, fleet_name: str, update_run_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateRuns/{updateRunName}/start" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateRunName": _SERIALIZER.url("update_run_name", update_run_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, **kwargs) def build_update_runs_stop_request( resource_group_name: str, fleet_name: str, update_run_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateRuns/{updateRunName}/stop" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateRunName": _SERIALIZER.url("update_run_name", update_run_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, **kwargs) def build_update_runs_skip_request( resource_group_name: str, fleet_name: str, update_run_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateRuns/{updateRunName}/skip" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateRunName": _SERIALIZER.url("update_run_name", update_run_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, **kwargs) def build_fleet_update_strategies_get_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, update_strategy_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateStrategies/{updateStrategyName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateStrategyName": _SERIALIZER.url("update_strategy_name", update_strategy_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_fleet_update_strategies_create_or_update_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, update_strategy_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateStrategies/{updateStrategyName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateStrategyName": _SERIALIZER.url("update_strategy_name", update_strategy_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs) def build_fleet_update_strategies_delete_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, update_strategy_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateStrategies/{updateStrategyName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "updateStrategyName": _SERIALIZER.url("update_strategy_name", update_strategy_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs) def build_fleet_update_strategies_list_by_fleet_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/updateStrategies" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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_auto_upgrade_profiles_get_request( resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/autoUpgradeProfiles/{autoUpgradeProfileName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "autoUpgradeProfileName": _SERIALIZER.url("auto_upgrade_profile_name", auto_upgrade_profile_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_auto_upgrade_profiles_create_or_update_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/autoUpgradeProfiles/{autoUpgradeProfileName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "autoUpgradeProfileName": _SERIALIZER.url("auto_upgrade_profile_name", auto_upgrade_profile_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs) def build_auto_upgrade_profiles_delete_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, subscription_id: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/autoUpgradeProfiles/{autoUpgradeProfileName}" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "autoUpgradeProfileName": _SERIALIZER.url("auto_upgrade_profile_name", auto_upgrade_profile_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") if_match = prep_if_match(etag, match_condition) if if_match is not None: _headers["If-Match"] = _SERIALIZER.header("if_match", if_match, "str") if_none_match = prep_if_none_match(etag, match_condition) if if_none_match is not None: _headers["If-None-Match"] = _SERIALIZER.header("if_none_match", if_none_match, "str") return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs) def build_auto_upgrade_profiles_list_by_fleet_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/autoUpgradeProfiles" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_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_auto_upgrade_profile_operations_generate_update_run_request( # pylint: disable=name-too-long resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, subscription_id: str, **kwargs: Any ) -> HttpRequest: _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = case_insensitive_dict(kwargs.pop("params", {}) or {}) api_version: str = kwargs.pop("api_version", _params.pop("api-version", "2025-04-01-preview")) accept = _headers.pop("Accept", "application/json") # Construct URL _url = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerService/fleets/{fleetName}/autoUpgradeProfiles/{autoUpgradeProfileName}/generateUpdateRun" path_format_arguments = { "subscriptionId": _SERIALIZER.url("subscription_id", subscription_id, "str"), "resourceGroupName": _SERIALIZER.url("resource_group_name", resource_group_name, "str"), "fleetName": _SERIALIZER.url("fleet_name", fleet_name, "str"), "autoUpgradeProfileName": _SERIALIZER.url("auto_upgrade_profile_name", auto_upgrade_profile_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) class Operations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.containerservicefleet.ContainerServiceFleetMgmtClient`'s :attr:`operations` attribute. """ def __init__(self, *args, **kwargs) -> None: input_args = list(args) self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client") self._config: ContainerServiceFleetMgmtClientConfiguration = ( input_args.pop(0) if input_args else kwargs.pop("config") ) self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace def list(self, **kwargs: Any) -> ItemPaged["_models.Operation"]: """List the operations for the provider. :return: An iterator like instance of Operation :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.containerservicefleet.models.Operation] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.Operation]] = kwargs.pop("cls", None) error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) def prepare_request(next_link=None): if not next_link: _request = build_operations_list_request( api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) else: # make call to next link with the client's api-version _parsed_next_link = urllib.parse.urlparse(next_link) _next_request_params = case_insensitive_dict( { key: [urllib.parse.quote(v) for v in value] for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items() } ) _next_request_params["api-version"] = self._config.api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) return _request def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize(List[_models.Operation], deserialized.get("value", [])) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data) class FleetsOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.containerservicefleet.ContainerServiceFleetMgmtClient`'s :attr:`fleets` attribute. """ def __init__(self, *args, **kwargs) -> None: input_args = list(args) self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client") self._config: ContainerServiceFleetMgmtClientConfiguration = ( input_args.pop(0) if input_args else kwargs.pop("config") ) self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace def get(self, resource_group_name: str, fleet_name: str, **kwargs: Any) -> _models.Fleet: """Gets a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :return: Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.mgmt.containerservicefleet.models.Fleet :raises ~azure.core.exceptions.HttpResponseError: """ error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.Fleet] = kwargs.pop("cls", None) _request = build_fleets_get_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.Fleet, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore def _begin_create_or_update_initial( self, resource_group_name: str, fleet_name: str, resource: Union[_models.Fleet, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(resource, (IOBase, bytes)): _content = resource else: _content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_fleets_begin_create_or_update_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 201: response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, resource: _models.Fleet, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Fleet]: """Creates or updates a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param resource: Resource create parameters. Required. :type resource: ~azure.mgmt.containerservicefleet.models.Fleet :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, resource: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Fleet]: """Creates or updates a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param resource: Resource create parameters. Required. :type resource: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, resource: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Fleet]: """Creates or updates a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param resource: Resource create parameters. Required. :type resource: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_create_or_update( self, resource_group_name: str, fleet_name: str, resource: Union[_models.Fleet, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Fleet]: """Creates or updates a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param resource: Resource create parameters. Is one of the following types: Fleet, JSON, IO[bytes] Required. :type resource: ~azure.mgmt.containerservicefleet.models.Fleet or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.Fleet] = 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._begin_create_or_update_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, resource=resource, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response = pipeline_response.http_response deserialized = _deserialize(_models.Fleet, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.Fleet].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.Fleet]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @api_version_validation( method_added_on="2023-06-15-preview", params_added_on={ "2023-06-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _begin_update_initial( self, resource_group_name: str, fleet_name: str, properties: Union[_models.FleetPatch, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(properties, (IOBase, bytes)): _content = properties else: _content = json.dumps(properties, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_fleets_begin_update_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_update( self, resource_group_name: str, fleet_name: str, properties: _models.FleetPatch, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Fleet]: """Update a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param properties: The resource properties to be updated. Required. :type properties: ~azure.mgmt.containerservicefleet.models.FleetPatch :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, fleet_name: str, properties: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Fleet]: """Update a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param properties: The resource properties to be updated. Required. :type properties: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, fleet_name: str, properties: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Fleet]: """Update a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param properties: The resource properties to be updated. Required. :type properties: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace @api_version_validation( method_added_on="2023-06-15-preview", params_added_on={ "2023-06-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_update( self, resource_group_name: str, fleet_name: str, properties: Union[_models.FleetPatch, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Fleet]: """Update a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param properties: The resource properties to be updated. Is one of the following types: FleetPatch, JSON, IO[bytes] Required. :type properties: ~azure.mgmt.containerservicefleet.models.FleetPatch or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Fleet. The Fleet is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.Fleet] = 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._begin_update_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, properties=properties, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response = pipeline_response.http_response deserialized = _deserialize(_models.Fleet, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.Fleet].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.Fleet]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) def _delete_initial( self, resource_group_name: str, fleet_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_fleets_delete_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace def begin_delete( self, resource_group_name: str, fleet_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[None]: """Delete a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns None :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[None] = kwargs.pop("cls", None) polling: Union[bool, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._delete_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, etag=etag, match_condition=match_condition, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements if cls: return cls(pipeline_response, None, {}) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[None].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[None](self._client, raw_result, get_long_running_output, polling_method) # type: ignore @distributed_trace def list_by_resource_group(self, resource_group_name: str, **kwargs: Any) -> ItemPaged["_models.Fleet"]: """Lists fleets in the specified subscription and resource group. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :return: An iterator like instance of Fleet :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.Fleet]] = 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_fleets_list_by_resource_group_request( resource_group_name=resource_group_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) else: # make call to next link with the client's api-version _parsed_next_link = urllib.parse.urlparse(next_link) _next_request_params = case_insensitive_dict( { key: [urllib.parse.quote(v) for v in value] for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items() } ) _next_request_params["api-version"] = self._config.api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) return _request def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize(List[_models.Fleet], deserialized.get("value", [])) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data) @distributed_trace def list_by_subscription(self, **kwargs: Any) -> ItemPaged["_models.Fleet"]: """Lists fleets in the specified subscription. :return: An iterator like instance of Fleet :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.containerservicefleet.models.Fleet] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.Fleet]] = 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_fleets_list_by_subscription_request( subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) else: # make call to next link with the client's api-version _parsed_next_link = urllib.parse.urlparse(next_link) _next_request_params = case_insensitive_dict( { key: [urllib.parse.quote(v) for v in value] for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items() } ) _next_request_params["api-version"] = self._config.api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) return _request def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize(List[_models.Fleet], deserialized.get("value", [])) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data) @distributed_trace def list_credentials( self, resource_group_name: str, fleet_name: str, **kwargs: Any ) -> _models.FleetCredentialResults: """Lists the user credentials of a Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :return: FleetCredentialResults. The FleetCredentialResults is compatible with MutableMapping :rtype: ~azure.mgmt.containerservicefleet.models.FleetCredentialResults :raises ~azure.core.exceptions.HttpResponseError: """ error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.FleetCredentialResults] = kwargs.pop("cls", None) _request = build_fleets_list_credentials_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.FleetCredentialResults, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore class FleetMembersOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.containerservicefleet.ContainerServiceFleetMgmtClient`'s :attr:`fleet_members` attribute. """ def __init__(self, *args, **kwargs) -> None: input_args = list(args) self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client") self._config: ContainerServiceFleetMgmtClientConfiguration = ( input_args.pop(0) if input_args else kwargs.pop("config") ) self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace def get( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, **kwargs: Any ) -> _models.FleetMember: """Get a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :return: FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.mgmt.containerservicefleet.models.FleetMember :raises ~azure.core.exceptions.HttpResponseError: """ error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.FleetMember] = kwargs.pop("cls", None) _request = build_fleet_members_get_request( resource_group_name=resource_group_name, fleet_name=fleet_name, fleet_member_name=fleet_member_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.FleetMember, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore def _create_initial( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, resource: Union[_models.FleetMember, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(resource, (IOBase, bytes)): _content = resource else: _content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_fleet_members_create_request( resource_group_name=resource_group_name, fleet_name=fleet_name, fleet_member_name=fleet_member_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 201: response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_create( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, resource: _models.FleetMember, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetMember]: """Create a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :param resource: Resource create parameters. Required. :type resource: ~azure.mgmt.containerservicefleet.models.FleetMember :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, resource: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetMember]: """Create a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :param resource: Resource create parameters. Required. :type resource: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, resource: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetMember]: """Create a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :param resource: Resource create parameters. Required. :type resource: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace def begin_create( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, resource: Union[_models.FleetMember, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetMember]: """Create a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :param resource: Resource create parameters. Is one of the following types: FleetMember, JSON, IO[bytes] Required. :type resource: ~azure.mgmt.containerservicefleet.models.FleetMember or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.FleetMember] = 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_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, fleet_member_name=fleet_member_name, resource=resource, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response = pipeline_response.http_response deserialized = _deserialize(_models.FleetMember, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.FleetMember].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.FleetMember]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @api_version_validation( method_added_on="2023-06-15-preview", params_added_on={ "2023-06-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "fleet_member_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _begin_update_initial( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, properties: Union[_models.FleetMemberUpdate, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(properties, (IOBase, bytes)): _content = properties else: _content = json.dumps(properties, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_fleet_members_begin_update_request( resource_group_name=resource_group_name, fleet_name=fleet_name, fleet_member_name=fleet_member_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_update( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, properties: _models.FleetMemberUpdate, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetMember]: """Update a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :param properties: The resource properties to be updated. Required. :type properties: ~azure.mgmt.containerservicefleet.models.FleetMemberUpdate :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, properties: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetMember]: """Update a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :param properties: The resource properties to be updated. Required. :type properties: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, properties: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetMember]: """Update a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :param properties: The resource properties to be updated. Required. :type properties: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace @api_version_validation( method_added_on="2023-06-15-preview", params_added_on={ "2023-06-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "fleet_member_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_update( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, properties: Union[_models.FleetMemberUpdate, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetMember]: """Update a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :param properties: The resource properties to be updated. Is one of the following types: FleetMemberUpdate, JSON, IO[bytes] Required. :type properties: ~azure.mgmt.containerservicefleet.models.FleetMemberUpdate or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetMember. The FleetMember is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.FleetMember] = 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._begin_update_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, fleet_member_name=fleet_member_name, properties=properties, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response = pipeline_response.http_response deserialized = _deserialize(_models.FleetMember, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.FleetMember].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.FleetMember]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) def _delete_initial( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_fleet_members_delete_request( resource_group_name=resource_group_name, fleet_name=fleet_name, fleet_member_name=fleet_member_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace def begin_delete( self, resource_group_name: str, fleet_name: str, fleet_member_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[None]: """Delete a FleetMember. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param fleet_member_name: The name of the Fleet member resource. Required. :type fleet_member_name: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns None :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[None] = kwargs.pop("cls", None) polling: Union[bool, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._delete_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, fleet_member_name=fleet_member_name, etag=etag, match_condition=match_condition, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements if cls: return cls(pipeline_response, None, {}) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[None].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[None](self._client, raw_result, get_long_running_output, polling_method) # type: ignore @distributed_trace def list_by_fleet( self, resource_group_name: str, fleet_name: str, **kwargs: Any ) -> ItemPaged["_models.FleetMember"]: """List FleetMember resources by Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :return: An iterator like instance of FleetMember :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.containerservicefleet.models.FleetMember] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.FleetMember]] = 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_fleet_members_list_by_fleet_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) else: # make call to next link with the client's api-version _parsed_next_link = urllib.parse.urlparse(next_link) _next_request_params = case_insensitive_dict( { key: [urllib.parse.quote(v) for v in value] for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items() } ) _next_request_params["api-version"] = self._config.api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) return _request def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize(List[_models.FleetMember], deserialized.get("value", [])) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data) class GatesOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.containerservicefleet.ContainerServiceFleetMgmtClient`'s :attr:`gates` attribute. """ def __init__(self, *args, **kwargs) -> None: input_args = list(args) self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client") self._config: ContainerServiceFleetMgmtClientConfiguration = ( input_args.pop(0) if input_args else kwargs.pop("config") ) self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace @api_version_validation( method_added_on="2025-04-01-preview", params_added_on={ "2025-04-01-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "gate_name", "accept", ] }, api_versions_list=["2025-04-01-preview"], ) def get(self, resource_group_name: str, fleet_name: str, gate_name: str, **kwargs: Any) -> _models.Gate: """Get a Gate. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param gate_name: The name of the Gate resource, a GUID. Required. :type gate_name: str :return: Gate. The Gate is compatible with MutableMapping :rtype: ~azure.mgmt.containerservicefleet.models.Gate :raises ~azure.core.exceptions.HttpResponseError: """ error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.Gate] = kwargs.pop("cls", None) _request = build_gates_get_request( resource_group_name=resource_group_name, fleet_name=fleet_name, gate_name=gate_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.Gate, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @api_version_validation( method_added_on="2025-04-01-preview", params_added_on={ "2025-04-01-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "gate_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=["2025-04-01-preview"], ) def _update_initial( self, resource_group_name: str, fleet_name: str, gate_name: str, properties: Union[_models.GatePatch, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(properties, (IOBase, bytes)): _content = properties else: _content = json.dumps(properties, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_gates_update_request( resource_group_name=resource_group_name, fleet_name=fleet_name, gate_name=gate_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_update( self, resource_group_name: str, fleet_name: str, gate_name: str, properties: _models.GatePatch, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Gate]: """Update a Gate. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param gate_name: The name of the Gate resource, a GUID. Required. :type gate_name: str :param properties: The resource properties to be updated. Required. :type properties: ~azure.mgmt.containerservicefleet.models.GatePatch :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Gate. The Gate is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Gate] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, fleet_name: str, gate_name: str, properties: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Gate]: """Update a Gate. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param gate_name: The name of the Gate resource, a GUID. Required. :type gate_name: str :param properties: The resource properties to be updated. Required. :type properties: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Gate. The Gate is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Gate] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_update( self, resource_group_name: str, fleet_name: str, gate_name: str, properties: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Gate]: """Update a Gate. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param gate_name: The name of the Gate resource, a GUID. Required. :type gate_name: str :param properties: The resource properties to be updated. Required. :type properties: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Gate. The Gate is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Gate] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace @api_version_validation( method_added_on="2025-04-01-preview", params_added_on={ "2025-04-01-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "gate_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=["2025-04-01-preview"], ) def begin_update( self, resource_group_name: str, fleet_name: str, gate_name: str, properties: Union[_models.GatePatch, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.Gate]: """Update a Gate. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param gate_name: The name of the Gate resource, a GUID. Required. :type gate_name: str :param properties: The resource properties to be updated. Is one of the following types: GatePatch, JSON, IO[bytes] Required. :type properties: ~azure.mgmt.containerservicefleet.models.GatePatch or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns Gate. The Gate is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.Gate] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.Gate] = 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, fleet_name=fleet_name, gate_name=gate_name, properties=properties, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response = pipeline_response.http_response deserialized = _deserialize(_models.Gate, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.Gate].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.Gate]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @distributed_trace @api_version_validation( method_added_on="2025-04-01-preview", params_added_on={ "2025-04-01-preview": ["api_version", "subscription_id", "resource_group_name", "fleet_name", "accept"] }, api_versions_list=["2025-04-01-preview"], ) def list_by_fleet(self, resource_group_name: str, fleet_name: str, **kwargs: Any) -> ItemPaged["_models.Gate"]: """List Gate resources by Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :return: An iterator like instance of Gate :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.containerservicefleet.models.Gate] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.Gate]] = 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_gates_list_by_fleet_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) else: # make call to next link with the client's api-version _parsed_next_link = urllib.parse.urlparse(next_link) _next_request_params = case_insensitive_dict( { key: [urllib.parse.quote(v) for v in value] for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items() } ) _next_request_params["api-version"] = self._config.api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) return _request def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize(List[_models.Gate], deserialized.get("value", [])) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data) class UpdateRunsOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.containerservicefleet.ContainerServiceFleetMgmtClient`'s :attr:`update_runs` attribute. """ def __init__(self, *args, **kwargs) -> None: input_args = list(args) self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client") self._config: ContainerServiceFleetMgmtClientConfiguration = ( input_args.pop(0) if input_args else kwargs.pop("config") ) self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "accept", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def get(self, resource_group_name: str, fleet_name: str, update_run_name: str, **kwargs: Any) -> _models.UpdateRun: """Get a UpdateRun. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :return: UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.mgmt.containerservicefleet.models.UpdateRun :raises ~azure.core.exceptions.HttpResponseError: """ error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.UpdateRun] = kwargs.pop("cls", None) _request = build_update_runs_get_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.UpdateRun, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _create_or_update_initial( self, resource_group_name: str, fleet_name: str, update_run_name: str, resource: Union[_models.UpdateRun, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(resource, (IOBase, bytes)): _content = resource else: _content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_update_runs_create_or_update_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 201: response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, update_run_name: str, resource: _models.UpdateRun, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Create a UpdateRun. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :param resource: Resource create parameters. Required. :type resource: ~azure.mgmt.containerservicefleet.models.UpdateRun :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, update_run_name: str, resource: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Create a UpdateRun. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :param resource: Resource create parameters. Required. :type resource: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, update_run_name: str, resource: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Create a UpdateRun. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :param resource: Resource create parameters. Required. :type resource: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_create_or_update( self, resource_group_name: str, fleet_name: str, update_run_name: str, resource: Union[_models.UpdateRun, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Create a UpdateRun. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :param resource: Resource create parameters. Is one of the following types: UpdateRun, JSON, IO[bytes] Required. :type resource: ~azure.mgmt.containerservicefleet.models.UpdateRun or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.UpdateRun] = 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, fleet_name=fleet_name, update_run_name=update_run_name, resource=resource, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response = pipeline_response.http_response deserialized = _deserialize(_models.UpdateRun, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.UpdateRun].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.UpdateRun]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _delete_initial( self, resource_group_name: str, fleet_name: str, update_run_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_update_runs_delete_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_delete( self, resource_group_name: str, fleet_name: str, update_run_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[None]: """Delete a UpdateRun. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns None :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[None] = kwargs.pop("cls", None) polling: Union[bool, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._delete_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, etag=etag, match_condition=match_condition, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements if cls: return cls(pipeline_response, None, {}) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[None].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[None](self._client, raw_result, get_long_running_output, polling_method) # type: ignore @distributed_trace @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": ["api_version", "subscription_id", "resource_group_name", "fleet_name", "accept"] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def list_by_fleet(self, resource_group_name: str, fleet_name: str, **kwargs: Any) -> ItemPaged["_models.UpdateRun"]: """List UpdateRun resources by Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :return: An iterator like instance of UpdateRun :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.UpdateRun]] = 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_update_runs_list_by_fleet_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) else: # make call to next link with the client's api-version _parsed_next_link = urllib.parse.urlparse(next_link) _next_request_params = case_insensitive_dict( { key: [urllib.parse.quote(v) for v in value] for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items() } ) _next_request_params["api-version"] = self._config.api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) return _request def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize(List[_models.UpdateRun], deserialized.get("value", [])) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data) @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _start_initial( self, resource_group_name: str, fleet_name: str, update_run_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_update_runs_start_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_start( self, resource_group_name: str, fleet_name: str, update_run_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Starts an UpdateRun. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.UpdateRun] = 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._start_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, etag=etag, match_condition=match_condition, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response_headers = {} response = pipeline_response.http_response response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = _deserialize(_models.UpdateRun, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.UpdateRun].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.UpdateRun]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _stop_initial( self, resource_group_name: str, fleet_name: str, update_run_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_update_runs_stop_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace @api_version_validation( method_added_on="2023-03-15-preview", params_added_on={ "2023-03-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-03-15-preview", "2023-06-15-preview", "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_stop( self, resource_group_name: str, fleet_name: str, update_run_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Stops an UpdateRun. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.UpdateRun] = 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._stop_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, etag=etag, match_condition=match_condition, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response_headers = {} response = pipeline_response.http_response response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = _deserialize(_models.UpdateRun, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.UpdateRun].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.UpdateRun]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @api_version_validation( method_added_on="2024-02-02-preview", params_added_on={ "2024-02-02-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _skip_initial( self, resource_group_name: str, fleet_name: str, update_run_name: str, body: Union[_models.SkipProperties, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(body, (IOBase, bytes)): _content = body else: _content = json.dumps(body, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_update_runs_skip_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_skip( self, resource_group_name: str, fleet_name: str, update_run_name: str, body: _models.SkipProperties, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Skips one or a combination of member/group/stage/afterStageWait(s) of an update run. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :param body: The content of the action request. Required. :type body: ~azure.mgmt.containerservicefleet.models.SkipProperties :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_skip( self, resource_group_name: str, fleet_name: str, update_run_name: str, body: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Skips one or a combination of member/group/stage/afterStageWait(s) of an update run. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :param body: The content of the action request. Required. :type body: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_skip( self, resource_group_name: str, fleet_name: str, update_run_name: str, body: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Skips one or a combination of member/group/stage/afterStageWait(s) of an update run. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :param body: The content of the action request. Required. :type body: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace @api_version_validation( method_added_on="2024-02-02-preview", params_added_on={ "2024-02-02-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_run_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_skip( self, resource_group_name: str, fleet_name: str, update_run_name: str, body: Union[_models.SkipProperties, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.UpdateRun]: """Skips one or a combination of member/group/stage/afterStageWait(s) of an update run. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_run_name: The name of the UpdateRun resource. Required. :type update_run_name: str :param body: The content of the action request. Is one of the following types: SkipProperties, JSON, IO[bytes] Required. :type body: ~azure.mgmt.containerservicefleet.models.SkipProperties or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns UpdateRun. The UpdateRun is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.UpdateRun] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.UpdateRun] = 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._skip_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, update_run_name=update_run_name, body=body, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response_headers = {} response = pipeline_response.http_response response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = _deserialize(_models.UpdateRun, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.UpdateRun].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.UpdateRun]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) class FleetUpdateStrategiesOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.containerservicefleet.ContainerServiceFleetMgmtClient`'s :attr:`fleet_update_strategies` attribute. """ def __init__(self, *args, **kwargs) -> None: input_args = list(args) self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client") self._config: ContainerServiceFleetMgmtClientConfiguration = ( input_args.pop(0) if input_args else kwargs.pop("config") ) self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace @api_version_validation( method_added_on="2023-08-15-preview", params_added_on={ "2023-08-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_strategy_name", "accept", ] }, api_versions_list=[ "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def get( self, resource_group_name: str, fleet_name: str, update_strategy_name: str, **kwargs: Any ) -> _models.FleetUpdateStrategy: """Get a FleetUpdateStrategy. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_strategy_name: The name of the UpdateStrategy resource. Required. :type update_strategy_name: str :return: FleetUpdateStrategy. The FleetUpdateStrategy is compatible with MutableMapping :rtype: ~azure.mgmt.containerservicefleet.models.FleetUpdateStrategy :raises ~azure.core.exceptions.HttpResponseError: """ error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.FleetUpdateStrategy] = kwargs.pop("cls", None) _request = build_fleet_update_strategies_get_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_strategy_name=update_strategy_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.FleetUpdateStrategy, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @api_version_validation( method_added_on="2023-08-15-preview", params_added_on={ "2023-08-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_strategy_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _create_or_update_initial( self, resource_group_name: str, fleet_name: str, update_strategy_name: str, resource: Union[_models.FleetUpdateStrategy, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(resource, (IOBase, bytes)): _content = resource else: _content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_fleet_update_strategies_create_or_update_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_strategy_name=update_strategy_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 201: response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, update_strategy_name: str, resource: _models.FleetUpdateStrategy, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetUpdateStrategy]: """Create a FleetUpdateStrategy. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_strategy_name: The name of the UpdateStrategy resource. Required. :type update_strategy_name: str :param resource: Resource create parameters. Required. :type resource: ~azure.mgmt.containerservicefleet.models.FleetUpdateStrategy :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetUpdateStrategy. The FleetUpdateStrategy is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetUpdateStrategy] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, update_strategy_name: str, resource: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetUpdateStrategy]: """Create a FleetUpdateStrategy. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_strategy_name: The name of the UpdateStrategy resource. Required. :type update_strategy_name: str :param resource: Resource create parameters. Required. :type resource: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetUpdateStrategy. The FleetUpdateStrategy is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetUpdateStrategy] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, update_strategy_name: str, resource: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetUpdateStrategy]: """Create a FleetUpdateStrategy. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_strategy_name: The name of the UpdateStrategy resource. Required. :type update_strategy_name: str :param resource: Resource create parameters. Required. :type resource: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetUpdateStrategy. The FleetUpdateStrategy is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetUpdateStrategy] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace @api_version_validation( method_added_on="2023-08-15-preview", params_added_on={ "2023-08-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_strategy_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_create_or_update( self, resource_group_name: str, fleet_name: str, update_strategy_name: str, resource: Union[_models.FleetUpdateStrategy, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.FleetUpdateStrategy]: """Create a FleetUpdateStrategy. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_strategy_name: The name of the UpdateStrategy resource. Required. :type update_strategy_name: str :param resource: Resource create parameters. Is one of the following types: FleetUpdateStrategy, JSON, IO[bytes] Required. :type resource: ~azure.mgmt.containerservicefleet.models.FleetUpdateStrategy or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns FleetUpdateStrategy. The FleetUpdateStrategy is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.FleetUpdateStrategy] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.FleetUpdateStrategy] = 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, fleet_name=fleet_name, update_strategy_name=update_strategy_name, resource=resource, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response = pipeline_response.http_response deserialized = _deserialize(_models.FleetUpdateStrategy, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.FleetUpdateStrategy].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.FleetUpdateStrategy]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @api_version_validation( method_added_on="2023-08-15-preview", params_added_on={ "2023-08-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_strategy_name", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def _delete_initial( self, resource_group_name: str, fleet_name: str, update_strategy_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_fleet_update_strategies_delete_request( resource_group_name=resource_group_name, fleet_name=fleet_name, update_strategy_name=update_strategy_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace @api_version_validation( method_added_on="2023-08-15-preview", params_added_on={ "2023-08-15-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "update_strategy_name", "accept", "etag", "match_condition", ] }, api_versions_list=[ "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def begin_delete( self, resource_group_name: str, fleet_name: str, update_strategy_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[None]: """Delete a FleetUpdateStrategy. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param update_strategy_name: The name of the UpdateStrategy resource. Required. :type update_strategy_name: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns None :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[None] = kwargs.pop("cls", None) polling: Union[bool, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._delete_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, update_strategy_name=update_strategy_name, etag=etag, match_condition=match_condition, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements if cls: return cls(pipeline_response, None, {}) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[None].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[None](self._client, raw_result, get_long_running_output, polling_method) # type: ignore @distributed_trace @api_version_validation( method_added_on="2023-08-15-preview", params_added_on={ "2023-08-15-preview": ["api_version", "subscription_id", "resource_group_name", "fleet_name", "accept"] }, api_versions_list=[ "2023-08-15-preview", "2023-10-15", "2024-02-02-preview", "2024-04-01", "2024-05-02-preview", "2025-03-01", "2025-04-01-preview", ], ) def list_by_fleet( self, resource_group_name: str, fleet_name: str, **kwargs: Any ) -> ItemPaged["_models.FleetUpdateStrategy"]: """List FleetUpdateStrategy resources by Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :return: An iterator like instance of FleetUpdateStrategy :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.containerservicefleet.models.FleetUpdateStrategy] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.FleetUpdateStrategy]] = 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_fleet_update_strategies_list_by_fleet_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) else: # make call to next link with the client's api-version _parsed_next_link = urllib.parse.urlparse(next_link) _next_request_params = case_insensitive_dict( { key: [urllib.parse.quote(v) for v in value] for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items() } ) _next_request_params["api-version"] = self._config.api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) return _request def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize(List[_models.FleetUpdateStrategy], deserialized.get("value", [])) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data) class AutoUpgradeProfilesOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.containerservicefleet.ContainerServiceFleetMgmtClient`'s :attr:`auto_upgrade_profiles` attribute. """ def __init__(self, *args, **kwargs) -> None: input_args = list(args) self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client") self._config: ContainerServiceFleetMgmtClientConfiguration = ( input_args.pop(0) if input_args else kwargs.pop("config") ) self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer") @distributed_trace @api_version_validation( method_added_on="2024-05-02-preview", params_added_on={ "2024-05-02-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "auto_upgrade_profile_name", "accept", ] }, api_versions_list=["2024-05-02-preview", "2025-03-01", "2025-04-01-preview"], ) def get( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, **kwargs: Any ) -> _models.AutoUpgradeProfile: """Get a AutoUpgradeProfile. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param auto_upgrade_profile_name: The name of the AutoUpgradeProfile resource. Required. :type auto_upgrade_profile_name: str :return: AutoUpgradeProfile. The AutoUpgradeProfile is compatible with MutableMapping :rtype: ~azure.mgmt.containerservicefleet.models.AutoUpgradeProfile :raises ~azure.core.exceptions.HttpResponseError: """ error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.AutoUpgradeProfile] = kwargs.pop("cls", None) _request = build_auto_upgrade_profiles_get_request( resource_group_name=resource_group_name, fleet_name=fleet_name, auto_upgrade_profile_name=auto_upgrade_profile_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = kwargs.pop("stream", False) pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: if _stream: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) if _stream: deserialized = response.iter_bytes() else: deserialized = _deserialize(_models.AutoUpgradeProfile, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized # type: ignore @api_version_validation( method_added_on="2024-05-02-preview", params_added_on={ "2024-05-02-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "auto_upgrade_profile_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=["2024-05-02-preview", "2025-03-01", "2025-04-01-preview"], ) def _create_or_update_initial( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, resource: Union[_models.AutoUpgradeProfile, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) content_type = content_type or "application/json" _content = None if isinstance(resource, (IOBase, bytes)): _content = resource else: _content = json.dumps(resource, cls=SdkJSONEncoder, exclude_readonly=True) # type: ignore _request = build_auto_upgrade_profiles_create_or_update_request( resource_group_name=resource_group_name, fleet_name=fleet_name, auto_upgrade_profile_name=auto_upgrade_profile_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, content_type=content_type, api_version=self._config.api_version, content=_content, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200, 201]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 201: response_headers["Azure-AsyncOperation"] = self._deserialize( "str", response.headers.get("Azure-AsyncOperation") ) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, resource: _models.AutoUpgradeProfile, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.AutoUpgradeProfile]: """Create a AutoUpgradeProfile. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param auto_upgrade_profile_name: The name of the AutoUpgradeProfile resource. Required. :type auto_upgrade_profile_name: str :param resource: Resource create parameters. Required. :type resource: ~azure.mgmt.containerservicefleet.models.AutoUpgradeProfile :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns AutoUpgradeProfile. The AutoUpgradeProfile is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.AutoUpgradeProfile] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, resource: JSON, *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.AutoUpgradeProfile]: """Create a AutoUpgradeProfile. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param auto_upgrade_profile_name: The name of the AutoUpgradeProfile resource. Required. :type auto_upgrade_profile_name: str :param resource: Resource create parameters. Required. :type resource: JSON :keyword content_type: Body Parameter content-type. Content type parameter for JSON body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns AutoUpgradeProfile. The AutoUpgradeProfile is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.AutoUpgradeProfile] :raises ~azure.core.exceptions.HttpResponseError: """ @overload def begin_create_or_update( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, resource: IO[bytes], *, content_type: str = "application/json", etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.AutoUpgradeProfile]: """Create a AutoUpgradeProfile. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param auto_upgrade_profile_name: The name of the AutoUpgradeProfile resource. Required. :type auto_upgrade_profile_name: str :param resource: Resource create parameters. Required. :type resource: IO[bytes] :keyword content_type: Body Parameter content-type. Content type parameter for binary body. Default value is "application/json". :paramtype content_type: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns AutoUpgradeProfile. The AutoUpgradeProfile is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.AutoUpgradeProfile] :raises ~azure.core.exceptions.HttpResponseError: """ @distributed_trace @api_version_validation( method_added_on="2024-05-02-preview", params_added_on={ "2024-05-02-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "auto_upgrade_profile_name", "content_type", "accept", "etag", "match_condition", ] }, api_versions_list=["2024-05-02-preview", "2025-03-01", "2025-04-01-preview"], ) def begin_create_or_update( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, resource: Union[_models.AutoUpgradeProfile, JSON, IO[bytes]], *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[_models.AutoUpgradeProfile]: """Create a AutoUpgradeProfile. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param auto_upgrade_profile_name: The name of the AutoUpgradeProfile resource. Required. :type auto_upgrade_profile_name: str :param resource: Resource create parameters. Is one of the following types: AutoUpgradeProfile, JSON, IO[bytes] Required. :type resource: ~azure.mgmt.containerservicefleet.models.AutoUpgradeProfile or JSON or IO[bytes] :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns AutoUpgradeProfile. The AutoUpgradeProfile is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.AutoUpgradeProfile] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = case_insensitive_dict(kwargs.pop("headers", {}) or {}) _params = kwargs.pop("params", {}) or {} content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None)) cls: ClsType[_models.AutoUpgradeProfile] = 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, fleet_name=fleet_name, auto_upgrade_profile_name=auto_upgrade_profile_name, resource=resource, etag=etag, match_condition=match_condition, content_type=content_type, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response = pipeline_response.http_response deserialized = _deserialize(_models.AutoUpgradeProfile, response.json()) if cls: return cls(pipeline_response, deserialized, {}) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.AutoUpgradeProfile].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.AutoUpgradeProfile]( self._client, raw_result, get_long_running_output, polling_method # type: ignore ) @api_version_validation( method_added_on="2024-05-02-preview", params_added_on={ "2024-05-02-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "auto_upgrade_profile_name", "accept", "etag", "match_condition", ] }, api_versions_list=["2024-05-02-preview", "2025-03-01", "2025-04-01-preview"], ) def _delete_initial( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } if match_condition == MatchConditions.IfNotModified: error_map[412] = ResourceModifiedError elif match_condition == MatchConditions.IfPresent: error_map[412] = ResourceNotFoundError elif match_condition == MatchConditions.IfMissing: error_map[412] = ResourceExistsError error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_auto_upgrade_profiles_delete_request( resource_group_name=resource_group_name, fleet_name=fleet_name, auto_upgrade_profile_name=auto_upgrade_profile_name, subscription_id=self._config.subscription_id, etag=etag, match_condition=match_condition, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [202, 204]: try: response.read() # Load the body in memory and close the socket except (StreamConsumedError, StreamClosedError): pass map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["Retry-After"] = self._deserialize("int", response.headers.get("Retry-After")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace @api_version_validation( method_added_on="2024-05-02-preview", params_added_on={ "2024-05-02-preview": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "auto_upgrade_profile_name", "accept", "etag", "match_condition", ] }, api_versions_list=["2024-05-02-preview", "2025-03-01", "2025-04-01-preview"], ) def begin_delete( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, *, etag: Optional[str] = None, match_condition: Optional[MatchConditions] = None, **kwargs: Any ) -> LROPoller[None]: """Delete a AutoUpgradeProfile. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param auto_upgrade_profile_name: The name of the AutoUpgradeProfile resource. Required. :type auto_upgrade_profile_name: str :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is None. :paramtype etag: str :keyword match_condition: The match condition to use upon the etag. Default value is None. :paramtype match_condition: ~azure.core.MatchConditions :return: An instance of LROPoller that returns None :rtype: ~azure.core.polling.LROPoller[None] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[None] = kwargs.pop("cls", None) polling: Union[bool, PollingMethod] = kwargs.pop("polling", True) lro_delay = kwargs.pop("polling_interval", self._config.polling_interval) cont_token: Optional[str] = kwargs.pop("continuation_token", None) if cont_token is None: raw_result = self._delete_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, auto_upgrade_profile_name=auto_upgrade_profile_name, etag=etag, match_condition=match_condition, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): # pylint: disable=inconsistent-return-statements if cls: return cls(pipeline_response, None, {}) # type: ignore path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[None].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[None](self._client, raw_result, get_long_running_output, polling_method) # type: ignore @distributed_trace @api_version_validation( method_added_on="2024-05-02-preview", params_added_on={ "2024-05-02-preview": ["api_version", "subscription_id", "resource_group_name", "fleet_name", "accept"] }, api_versions_list=["2024-05-02-preview", "2025-03-01", "2025-04-01-preview"], ) def list_by_fleet( self, resource_group_name: str, fleet_name: str, **kwargs: Any ) -> ItemPaged["_models.AutoUpgradeProfile"]: """List AutoUpgradeProfile resources by Fleet. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :return: An iterator like instance of AutoUpgradeProfile :rtype: ~azure.core.paging.ItemPaged[~azure.mgmt.containerservicefleet.models.AutoUpgradeProfile] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[List[_models.AutoUpgradeProfile]] = 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_auto_upgrade_profiles_list_by_fleet_request( resource_group_name=resource_group_name, fleet_name=fleet_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) else: # make call to next link with the client's api-version _parsed_next_link = urllib.parse.urlparse(next_link) _next_request_params = case_insensitive_dict( { key: [urllib.parse.quote(v) for v in value] for key, value in urllib.parse.parse_qs(_parsed_next_link.query).items() } ) _next_request_params["api-version"] = self._config.api_version _request = HttpRequest( "GET", urllib.parse.urljoin(next_link, _parsed_next_link.path), params=_next_request_params ) path_format_arguments = { "endpoint": self._serialize.url( "self._config.base_url", self._config.base_url, "str", skip_quote=True ), } _request.url = self._client.format_url(_request.url, **path_format_arguments) return _request def extract_data(pipeline_response): deserialized = pipeline_response.http_response.json() list_of_elem = _deserialize(List[_models.AutoUpgradeProfile], deserialized.get("value", [])) if cls: list_of_elem = cls(list_of_elem) # type: ignore return deserialized.get("nextLink") or None, iter(list_of_elem) def get_next(next_link=None): _request = prepare_request(next_link) _stream = False pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) return pipeline_response return ItemPaged(get_next, extract_data) class AutoUpgradeProfileOperationsOperations: """ .. warning:: **DO NOT** instantiate this class directly. Instead, you should access the following operations through :class:`~azure.mgmt.containerservicefleet.ContainerServiceFleetMgmtClient`'s :attr:`auto_upgrade_profile_operations` attribute. """ def __init__(self, *args, **kwargs) -> None: input_args = list(args) self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client") self._config: ContainerServiceFleetMgmtClientConfiguration = ( input_args.pop(0) if input_args else kwargs.pop("config") ) self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer") self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer") @api_version_validation( method_added_on="2025-03-01", params_added_on={ "2025-03-01": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "auto_upgrade_profile_name", "accept", ] }, api_versions_list=["2025-03-01", "2025-04-01-preview"], ) def _generate_update_run_initial( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, **kwargs: Any ) -> Iterator[bytes]: error_map: MutableMapping = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError, 304: ResourceNotModifiedError, } error_map.update(kwargs.pop("error_map", {}) or {}) _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None) _request = build_auto_upgrade_profile_operations_generate_update_run_request( resource_group_name=resource_group_name, fleet_name=fleet_name, auto_upgrade_profile_name=auto_upgrade_profile_name, subscription_id=self._config.subscription_id, api_version=self._config.api_version, headers=_headers, params=_params, ) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } _request.url = self._client.format_url(_request.url, **path_format_arguments) _stream = True pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access _request, stream=_stream, **kwargs ) response = pipeline_response.http_response if response.status_code not in [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) error = _failsafe_deserialize(_models.ErrorResponse, response.json()) raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) response_headers = {} if response.status_code == 202: response_headers["Azure-AsyncOperation"] = self._deserialize( "str", response.headers.get("Azure-AsyncOperation") ) response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["If-Match"] = self._deserialize("str", response.headers.get("If-Match")) deserialized = response.iter_bytes() if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized # type: ignore @distributed_trace @api_version_validation( method_added_on="2025-03-01", params_added_on={ "2025-03-01": [ "api_version", "subscription_id", "resource_group_name", "fleet_name", "auto_upgrade_profile_name", "accept", ] }, api_versions_list=["2025-03-01", "2025-04-01-preview"], ) def begin_generate_update_run( self, resource_group_name: str, fleet_name: str, auto_upgrade_profile_name: str, **kwargs: Any ) -> LROPoller[_models.GenerateResponse]: """Generates an update run for a given auto upgrade profile. :param resource_group_name: The name of the resource group. The name is case insensitive. Required. :type resource_group_name: str :param fleet_name: The name of the Fleet resource. Required. :type fleet_name: str :param auto_upgrade_profile_name: The name of the AutoUpgradeProfile resource. Required. :type auto_upgrade_profile_name: str :return: An instance of LROPoller that returns GenerateResponse. The GenerateResponse is compatible with MutableMapping :rtype: ~azure.core.polling.LROPoller[~azure.mgmt.containerservicefleet.models.GenerateResponse] :raises ~azure.core.exceptions.HttpResponseError: """ _headers = kwargs.pop("headers", {}) or {} _params = kwargs.pop("params", {}) or {} cls: ClsType[_models.GenerateResponse] = 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._generate_update_run_initial( resource_group_name=resource_group_name, fleet_name=fleet_name, auto_upgrade_profile_name=auto_upgrade_profile_name, cls=lambda x, y, z: x, headers=_headers, params=_params, **kwargs ) raw_result.http_response.read() # type: ignore kwargs.pop("error_map", None) def get_long_running_output(pipeline_response): response_headers = {} response = pipeline_response.http_response response_headers["Azure-AsyncOperation"] = self._deserialize( "str", response.headers.get("Azure-AsyncOperation") ) response_headers["Location"] = self._deserialize("str", response.headers.get("Location")) response_headers["If-Match"] = self._deserialize("str", response.headers.get("If-Match")) deserialized = _deserialize(_models.GenerateResponse, response.json()) if cls: return cls(pipeline_response, deserialized, response_headers) # type: ignore return deserialized path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } if polling is True: polling_method: PollingMethod = cast( PollingMethod, ARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) ) elif polling is False: polling_method = cast(PollingMethod, NoPolling()) else: polling_method = polling if cont_token: return LROPoller[_models.GenerateResponse].from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output, ) return LROPoller[_models.GenerateResponse]( self._client, raw_result, get_long_running_output, polling_method # type: ignore )