# coding=utf-8 # -------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for license information. # Code generated by Microsoft (R) Python Code Generator. # Changes may cause incorrect behavior and will be lost if the code is regenerated. # -------------------------------------------------------------------------- from copy import deepcopy from typing import Any, Awaitable, Optional, TYPE_CHECKING, cast from typing_extensions import Self from azure.core.pipeline import policies from azure.core.rest import AsyncHttpResponse, HttpRequest from azure.core.settings import settings from azure.mgmt.core import AsyncARMPipelineClient from azure.mgmt.core.policies import AsyncARMAutoResourceProviderRegistrationPolicy from azure.mgmt.core.tools import get_arm_endpoints from .._utils.serialization import Deserializer, Serializer from ._configuration import ContainerServiceFleetMgmtClientConfiguration from .operations import ( AutoUpgradeProfileOperationsOperations, AutoUpgradeProfilesOperations, FleetMembersOperations, FleetUpdateStrategiesOperations, FleetsOperations, GatesOperations, Operations, UpdateRunsOperations, ) if TYPE_CHECKING: from azure.core.credentials_async import AsyncTokenCredential class ContainerServiceFleetMgmtClient: # pylint: disable=too-many-instance-attributes """Azure Kubernetes Fleet Manager api client. :ivar operations: Operations operations :vartype operations: azure.mgmt.containerservicefleet.aio.operations.Operations :ivar fleets: FleetsOperations operations :vartype fleets: azure.mgmt.containerservicefleet.aio.operations.FleetsOperations :ivar fleet_members: FleetMembersOperations operations :vartype fleet_members: azure.mgmt.containerservicefleet.aio.operations.FleetMembersOperations :ivar gates: GatesOperations operations :vartype gates: azure.mgmt.containerservicefleet.aio.operations.GatesOperations :ivar update_runs: UpdateRunsOperations operations :vartype update_runs: azure.mgmt.containerservicefleet.aio.operations.UpdateRunsOperations :ivar fleet_update_strategies: FleetUpdateStrategiesOperations operations :vartype fleet_update_strategies: azure.mgmt.containerservicefleet.aio.operations.FleetUpdateStrategiesOperations :ivar auto_upgrade_profiles: AutoUpgradeProfilesOperations operations :vartype auto_upgrade_profiles: azure.mgmt.containerservicefleet.aio.operations.AutoUpgradeProfilesOperations :ivar auto_upgrade_profile_operations: AutoUpgradeProfileOperationsOperations operations :vartype auto_upgrade_profile_operations: azure.mgmt.containerservicefleet.aio.operations.AutoUpgradeProfileOperationsOperations :param credential: Credential used to authenticate requests to the service. Required. :type credential: ~azure.core.credentials_async.AsyncTokenCredential :param subscription_id: The ID of the target subscription. The value must be an UUID. Required. :type subscription_id: str :param base_url: Service host. Default value is None. :type base_url: str :keyword api_version: The API version to use for this operation. Default value is "2025-04-01-preview". Note that overriding this default value may result in unsupported behavior. :paramtype api_version: str :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. """ def __init__( self, credential: "AsyncTokenCredential", subscription_id: str, base_url: Optional[str] = None, **kwargs: Any ) -> None: _endpoint = "{endpoint}" _cloud = kwargs.pop("cloud_setting", None) or settings.current.azure_cloud # type: ignore _endpoints = get_arm_endpoints(_cloud) if not base_url: base_url = _endpoints["resource_manager"] credential_scopes = kwargs.pop("credential_scopes", _endpoints["credential_scopes"]) self._config = ContainerServiceFleetMgmtClientConfiguration( credential=credential, subscription_id=subscription_id, base_url=cast(str, base_url), credential_scopes=credential_scopes, **kwargs ) _policies = kwargs.pop("policies", None) if _policies is None: _policies = [ policies.RequestIdPolicy(**kwargs), self._config.headers_policy, self._config.user_agent_policy, self._config.proxy_policy, policies.ContentDecodePolicy(**kwargs), AsyncARMAutoResourceProviderRegistrationPolicy(), self._config.redirect_policy, self._config.retry_policy, self._config.authentication_policy, self._config.custom_hook_policy, self._config.logging_policy, policies.DistributedTracingPolicy(**kwargs), policies.SensitiveHeaderCleanupPolicy(**kwargs) if self._config.redirect_policy else None, self._config.http_logging_policy, ] self._client: AsyncARMPipelineClient = AsyncARMPipelineClient( base_url=cast(str, _endpoint), policies=_policies, **kwargs ) self._serialize = Serializer() self._deserialize = Deserializer() self._serialize.client_side_validation = False self.operations = Operations(self._client, self._config, self._serialize, self._deserialize) self.fleets = FleetsOperations(self._client, self._config, self._serialize, self._deserialize) self.fleet_members = FleetMembersOperations(self._client, self._config, self._serialize, self._deserialize) self.gates = GatesOperations(self._client, self._config, self._serialize, self._deserialize) self.update_runs = UpdateRunsOperations(self._client, self._config, self._serialize, self._deserialize) self.fleet_update_strategies = FleetUpdateStrategiesOperations( self._client, self._config, self._serialize, self._deserialize ) self.auto_upgrade_profiles = AutoUpgradeProfilesOperations( self._client, self._config, self._serialize, self._deserialize ) self.auto_upgrade_profile_operations = AutoUpgradeProfileOperationsOperations( self._client, self._config, self._serialize, self._deserialize ) def send_request( self, request: HttpRequest, *, stream: bool = False, **kwargs: Any ) -> Awaitable[AsyncHttpResponse]: """Runs the network request through the client's chained policies. >>> from azure.core.rest import HttpRequest >>> request = HttpRequest("GET", "https://www.example.org/") >>> response = await client.send_request(request) For more information on this code flow, see https://aka.ms/azsdk/dpcodegen/python/send_request :param request: The network request you want to make. Required. :type request: ~azure.core.rest.HttpRequest :keyword bool stream: Whether the response payload will be streamed. Defaults to False. :return: The response of your network call. Does not do error handling on your response. :rtype: ~azure.core.rest.AsyncHttpResponse """ request_copy = deepcopy(request) path_format_arguments = { "endpoint": self._serialize.url("self._config.base_url", self._config.base_url, "str", skip_quote=True), } request_copy.url = self._client.format_url(request_copy.url, **path_format_arguments) return self._client.send_request(request_copy, stream=stream, **kwargs) # type: ignore async def close(self) -> None: await self._client.close() async def __aenter__(self) -> Self: await self._client.__aenter__() return self async def __aexit__(self, *exc_details: Any) -> None: await self._client.__aexit__(*exc_details)