Class AsyncDstackClient

sdk/python/src/dstack_sdk/dstack_client.py:282–534

class AsyncDstackClient(BaseClient):
    PATH_PREFIX = "/"

    def __init__(
        self,
        endpoint: str | None = None,
        *,
        use_sync_http: bool = False,
        timeout: float = 3,
    ):
        """Initialize async client with HTTP or Unix-socket transport.

        Args:
            endpoint: HTTP/HTTPS URL or Unix socket path
            use_sync_http: If True, use sync HTTP client internally
            timeout: Timeout in seconds

        """
        endpoint = get_endpoint(endpoint)
        self.use_sync_http = use_sync_http
        self._client: Optional[httpx.AsyncClient] = None
        self._sync_client: Optional[httpx.Client] = None
        self._client_ref_count = 0
        self._timeout = timeout

        if endpoint.startswith("http://") or endpoint.startswith("https://"):
            self.async_transport = httpx.AsyncHTTPTransport()
            self.sync_transport = httpx.HTTPTransport()
            self.base_url = endpoint
        else:
            # Check if Unix socket file exists
            if endpoint.startswith("/") and not os.path.exists(endpoint):
                raise FileNotFoundError(f"Unix socket file {endpoint} does not exist")
            self.async_transport = httpx.AsyncHTTPTransport(uds=endpoint)
            self.sync_transport = httpx.HTTPTransport(uds=endpoint)
            self.base_url = "http://localhost"

    def _get_client(self) -> httpx.AsyncClient:
        if self._client is None:
            self._client = httpx.AsyncClient(
                transport=self.async_transport,
                base_url=self.base_url,
                timeout=self._timeout,
            )
        return self._client

    def _get_sync_client(self) -> httpx.Client:
        if self._sync_client is None:
            self._sync_client = httpx.Client(
                transport=self.sync_transport,
                base_url=self.base_url,
                timeout=self._timeout,
            )
        return self._sync_client

    async def _send_rpc_request(
        self, method: str, payload: Dict[str, Any]
    ) -> Dict[str, Any]: