44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
"""Worker instance management."""
|
|
|
|
from materia.providers import Instance, get_provider
|
|
from materia.secrets import get_secret
|
|
|
|
DEFAULT_PROVIDER = "hetzner"
|
|
|
|
|
|
def list_workers(provider: str = DEFAULT_PROVIDER) -> list[Instance]:
|
|
p = get_provider(provider)
|
|
return p.list_instances()
|
|
|
|
|
|
def create_worker(
|
|
name: str,
|
|
server_type: str,
|
|
provider: str = DEFAULT_PROVIDER,
|
|
location: str | None = None,
|
|
) -> Instance:
|
|
ssh_key = get_secret("SSH_PUBLIC_KEY")
|
|
if not ssh_key:
|
|
raise ValueError("SSH_PUBLIC_KEY not found in secrets")
|
|
|
|
p = get_provider(provider)
|
|
instance = p.create_instance(name, server_type, ssh_key, location)
|
|
|
|
if not p.wait_for_ssh(instance.ip):
|
|
raise RuntimeError(f"SSH never became available on {instance.ip}")
|
|
|
|
return instance
|
|
|
|
|
|
def destroy_worker(name: str, provider: str = DEFAULT_PROVIDER) -> None:
|
|
p = get_provider(provider)
|
|
instance = p.get_instance(name)
|
|
if not instance:
|
|
raise ValueError(f"Worker '{name}' not found")
|
|
p.destroy_instance(instance.id)
|
|
|
|
|
|
def get_worker(name: str, provider: str = DEFAULT_PROVIDER) -> Instance | None:
|
|
p = get_provider(provider)
|
|
return p.get_instance(name)
|