vultr-api/server/routers/instances.py

"""Instances router"""
from fastapi import APIRouter, Depends, Query
from typing import Optional, List
from pydantic import BaseModel

from vultr_api import VultrClient
from deps import get_client

router = APIRouter()


class CreateInstanceRequest(BaseModel):
    region: str
    plan: str
    os_id: Optional[int] = None
    iso_id: Optional[str] = None
    snapshot_id: Optional[str] = None
    app_id: Optional[int] = None
    image_id: Optional[str] = None
    script_id: Optional[str] = None
    ssh_key_ids: Optional[List[str]] = None
    backups: Optional[str] = None
    enable_ipv6: Optional[bool] = None
    hostname: Optional[str] = None
    label: Optional[str] = None
    tags: Optional[List[str]] = None
    firewall_group_id: Optional[str] = None
    vpc_id: Optional[str] = None


class UpdateInstanceRequest(BaseModel):
    label: Optional[str] = None
    tags: Optional[List[str]] = None
    firewall_group_id: Optional[str] = None
    enable_ipv6: Optional[bool] = None
    backups: Optional[str] = None


@router.get("")
async def list_instances(
    per_page: int = Query(25, le=500),
    cursor: Optional[str] = None,
    client: VultrClient = Depends(get_client)
):
    """List all instances"""
    return client.instances.list(per_page=per_page, cursor=cursor)


@router.get("/all")
async def list_all_instances(client: VultrClient = Depends(get_client)):
    """List all instances (auto-paginated)"""
    return {"instances": client.instances.list_all()}


@router.get("/{instance_id}")
async def get_instance(instance_id: str, client: VultrClient = Depends(get_client)):
    """Get instance details"""
    return client.instances.get(instance_id)


@router.post("")
async def create_instance(req: CreateInstanceRequest, client: VultrClient = Depends(get_client)):
    """Create a new instance"""
    return client.instances.create(**req.model_dump(exclude_none=True))


@router.patch("/{instance_id}")
async def update_instance(instance_id: str, req: UpdateInstanceRequest, client: VultrClient = Depends(get_client)):
    """Update an instance"""
    return client.instances.update(instance_id, **req.model_dump(exclude_none=True))


@router.delete("/{instance_id}")
async def delete_instance(instance_id: str, client: VultrClient = Depends(get_client)):
    """Delete an instance"""
    return client.instances.delete(instance_id)


@router.post("/{instance_id}/start")
async def start_instance(instance_id: str, client: VultrClient = Depends(get_client)):
    """Start an instance"""
    return client.instances.start(instance_id)


@router.post("/{instance_id}/stop")
async def stop_instance(instance_id: str, client: VultrClient = Depends(get_client)):
    """Stop an instance (halt)"""
    return client.instances.halt(instance_id)


@router.post("/{instance_id}/reboot")
async def reboot_instance(instance_id: str, client: VultrClient = Depends(get_client)):
    """Reboot an instance"""
    return client.instances.reboot(instance_id)


@router.post("/{instance_id}/reinstall")
async def reinstall_instance(instance_id: str, hostname: Optional[str] = None, client: VultrClient = Depends(get_client)):
    """Reinstall an instance"""
    return client.instances.reinstall(instance_id, hostname=hostname)


@router.get("/{instance_id}/ipv4")
async def list_ipv4(instance_id: str, client: VultrClient = Depends(get_client)):
    """List IPv4 addresses for an instance"""
    return client.instances.list_ipv4(instance_id)


@router.get("/{instance_id}/ipv6")
async def list_ipv6(instance_id: str, client: VultrClient = Depends(get_client)):
    """List IPv6 addresses for an instance"""
    return client.instances.list_ipv6(instance_id)


@router.get("/{instance_id}/bandwidth")
async def get_bandwidth(instance_id: str, client: VultrClient = Depends(get_client)):
    """Get bandwidth usage for an instance"""
    return client.instances.bandwidth(instance_id)


@router.get("/{instance_id}/neighbors")
async def get_neighbors(instance_id: str, client: VultrClient = Depends(get_client)):
    """Get instance neighbors"""
    return client.instances.neighbors(instance_id)


@router.get("/{instance_id}/user-data")
async def get_user_data(instance_id: str, client: VultrClient = Depends(get_client)):
    """Get instance user data"""
    return client.instances.get_user_data(instance_id)


@router.post("/{instance_id}/backup")
async def create_backup(instance_id: str, client: VultrClient = Depends(get_client)):
    """Create a backup of an instance"""
    return client.instances.create_backup(instance_id)


@router.post("/{instance_id}/restore")
async def restore_instance(
    instance_id: str,
    backup_id: Optional[str] = None,
    snapshot_id: Optional[str] = None,
    client: VultrClient = Depends(get_client)
):
    """Restore an instance from backup or snapshot"""
    if backup_id:
        return client.instances.restore_backup(instance_id, backup_id)
    elif snapshot_id:
        return client.instances.restore_snapshot(instance_id, snapshot_id)
    return {"error": "backup_id or snapshot_id required"}