added shared provider vm type

This commit is contained in:
Dillon DuPont
2025-10-09 12:36:43 -04:00
parent 844f138881
commit b29f89597f
5 changed files with 152 additions and 19 deletions

View File

@@ -2,7 +2,9 @@
import abc
from enum import StrEnum
from typing import Dict, List, Optional, Any, AsyncContextManager
from typing import Dict, Optional, Any, AsyncContextManager
from .types import ListVMsResponse
class VMProviderType(StrEnum):
@@ -42,8 +44,13 @@ class BaseVMProvider(AsyncContextManager):
pass
@abc.abstractmethod
async def list_vms(self) -> List[Dict[str, Any]]:
"""List all available VMs."""
async def list_vms(self) -> ListVMsResponse:
"""List all available VMs.
Returns:
ListVMsResponse: A list of minimal VM objects as defined in
`computer.providers.types.MinimalVM`.
"""
pass
@abc.abstractmethod
@@ -76,6 +83,20 @@ class BaseVMProvider(AsyncContextManager):
"""
pass
@abc.abstractmethod
async def restart_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
"""Restart a VM by name.
Args:
name: Name of the VM to restart
storage: Optional storage path override. If provided, this will be used
instead of the provider's default storage path.
Returns:
Dictionary with VM restart status and information
"""
pass
@abc.abstractmethod
async def update_vm(self, name: str, update_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any]:
"""Update VM configuration.

View File

@@ -12,6 +12,7 @@ import logging
from typing import Dict, List, Optional, Any
from ..base import BaseVMProvider, VMProviderType
from ..types import ListVMsResponse, MinimalVM
# Setup logging
logger = logging.getLogger(__name__)
@@ -106,7 +107,7 @@ class CloudProvider(BaseVMProvider):
# Host does not resolve → not found
return {"name": name, "status": "not_found", "hostname": hostname}
async def list_vms(self) -> List[Dict[str, Any]]:
async def list_vms(self) -> ListVMsResponse:
url = f"{self.api_base}/v1/vms"
headers = {
"Authorization": f"Bearer {self.api_key}",
@@ -122,7 +123,24 @@ class CloudProvider(BaseVMProvider):
logger.error(f"Failed to parse list_vms JSON: {text}")
return []
if isinstance(data, list):
return data
# Enrich with convenience URLs when possible.
enriched: List[Dict[str, Any]] = []
for item in data:
vm = dict(item) if isinstance(item, dict) else {}
name = vm.get("name")
password = vm.get("password")
if isinstance(name, str) and name:
host = f"https://{name}.containers.cloud.trycua.com:8443"
# api_url: always set if missing
if not vm.get("api_url"):
vm["api_url"] = host
# vnc_url: only when password available
if not vm.get("vnc_url") and isinstance(password, str) and password:
vm[
"vnc_url"
] = f"https://{host}/vnc.html?autoconnect=true&password={password}"
enriched.append(vm)
return enriched # type: ignore[return-value]
logger.warning("Unexpected response for list_vms; expected list")
return []
elif resp.status == 401:

View File

@@ -0,0 +1,57 @@
"""Pydantic models for the CUA Cloud provider API.
Documents the response shape for the Cloud list VMs endpoint.
List VMs
- Method: GET
- Path: `/v1/vms`
- Description: Returns all VMs owned by the API key's user.
- Responses:
- 200: Array of minimal VM objects with fields `{ name, password, status }`
- 401: Unauthorized (missing/invalid API key)
Example curl:
curl -H "Authorization: Bearer $CUA_API_KEY" \
"https://api.cua.ai/v1/vms"
Response shape:
[
{
"name": "s-windows-x4snp46ebf",
"password": "49b8daa3",
"status": "running"
}
]
Status values:
- pending : VM deployment in progress
- running : VM is active and accessible
- stopped : VM is stopped but not terminated
- terminated: VM has been permanently destroyed
- failed : VM deployment or operation failed
"""
from __future__ import annotations
from typing import Literal, Optional
# Require pydantic for typed models in provider APIs
from pydantic import BaseModel
CloudVMStatus = Literal["pending", "running", "stopped", "terminated", "failed"]
class CloudVM(BaseModel):
"""Minimal VM object returned by CUA Cloud list API.
Additional optional fields (like URLs) may be filled by callers based on
their environment but are not guaranteed by the API.
"""
name: str
password: str
status: CloudVMStatus
# Optional, not guaranteed by the list API, but useful when known/derived
vnc_url: Optional[str] = None
api_url: Optional[str] = None

View File

@@ -0,0 +1,35 @@
"""Shared provider type definitions for VM metadata and responses.
These base types describe the common shape of objects returned by provider
methods like `list_vms()`.
"""
from __future__ import annotations
from typing import Literal, TypedDict, NotRequired
# Core status values per product docs
VMStatus = Literal[
"pending", # VM deployment in progress
"running", # VM is active and accessible
"stopped", # VM is stopped but not terminated
"terminated", # VM has been permanently destroyed
"failed", # VM deployment or operation failed
]
class MinimalVM(TypedDict):
"""Minimal VM object shape returned by list calls.
Providers may include additional fields. Optional fields below are
common extensions some providers expose or that callers may compute.
"""
name: str
status: VMStatus
# Not always included by all providers
password: NotRequired[str]
vnc_url: NotRequired[str]
api_url: NotRequired[str]
# Convenience alias for list_vms() responses
ListVMsResponse = list[MinimalVM]