vultr-api/vultr_api/resources/instances.py

"""
Instances Resource

Cloud compute instance management
"""

from typing import Dict, List, Optional
from .base import BaseResource


class InstancesResource(BaseResource):
    """
    Cloud compute instances management

    Usage:
        # List all instances
        instances = client.instances.list()

        # Get specific instance
        instance = client.instances.get("instance-id")

        # Create instance
        instance = client.instances.create(
            region="ewr",
            plan="vc2-1c-1gb",
            os_id=215,
            label="my-server"
        )

        # Delete instance
        client.instances.delete("instance-id")
    """

    def list(
        self,
        per_page: int = 100,
        cursor: str = None,
        tag: str = None,
        label: str = None,
        main_ip: str = None,
        region: str = None,
    ) -> Dict:
        """
        List all instances

        Args:
            per_page: Number of items per page (max 500)
            cursor: Pagination cursor
            tag: Filter by tag
            label: Filter by label
            main_ip: Filter by main IP
            region: Filter by region

        Returns:
            Dict with 'instances' list and 'meta' pagination info
        """
        params = {"per_page": per_page}
        if cursor:
            params["cursor"] = cursor
        if tag:
            params["tag"] = tag
        if label:
            params["label"] = label
        if main_ip:
            params["main_ip"] = main_ip
        if region:
            params["region"] = region

        return self.client.get("instances", params=params)

    def list_all(self, **kwargs) -> List[Dict]:
        """
        List all instances (auto-paginate)

        Returns:
            List of all instances
        """
        return self.client.paginate("instances", "instances", params=kwargs)

    def get(self, instance_id: str) -> Dict:
        """
        Get instance details

        Args:
            instance_id: Instance ID

        Returns:
            Instance details
        """
        response = self.client.get(f"instances/{instance_id}")
        return response.get("instance", {})

    def create(
        self,
        region: str,
        plan: str,
        os_id: int = None,
        iso_id: str = None,
        snapshot_id: str = None,
        app_id: int = None,
        image_id: str = None,
        ipxe_chain_url: str = None,
        script_id: str = None,
        enable_ipv6: bool = False,
        disable_public_ipv4: bool = False,
        attach_private_network: List[str] = None,
        attach_vpc: List[str] = None,
        attach_vpc2: List[str] = None,
        label: str = None,
        sshkey_id: List[str] = None,
        backups: str = None,
        ddos_protection: bool = False,
        activation_email: bool = False,
        hostname: str = None,
        tag: str = None,
        tags: List[str] = None,
        firewall_group_id: str = None,
        reserved_ipv4: str = None,
        user_data: str = None,
    ) -> Dict:
        """
        Create a new instance

        Args:
            region: Region ID (e.g., "ewr", "lax", "nrt")
            plan: Plan ID (e.g., "vc2-1c-1gb")
            os_id: Operating System ID
            iso_id: ISO ID for custom install
            snapshot_id: Snapshot ID to restore from
            app_id: Application ID
            image_id: Custom image ID
            ipxe_chain_url: iPXE chain URL
            script_id: Startup script ID
            enable_ipv6: Enable IPv6
            disable_public_ipv4: Disable public IPv4
            attach_private_network: Private network IDs to attach
            attach_vpc: VPC IDs to attach
            attach_vpc2: VPC 2.0 IDs to attach
            label: Instance label
            sshkey_id: SSH key IDs to add
            backups: Backup schedule ("enabled" or "disabled")
            ddos_protection: Enable DDoS protection
            activation_email: Send activation email
            hostname: Server hostname
            tag: Deprecated, use tags
            tags: Tags for the instance
            firewall_group_id: Firewall group ID
            reserved_ipv4: Reserved IPv4 to assign
            user_data: Cloud-init user data (base64)

        Returns:
            Created instance details
        """
        data = {"region": region, "plan": plan}

        # Add optional parameters
        if os_id is not None:
            data["os_id"] = os_id
        if iso_id:
            data["iso_id"] = iso_id
        if snapshot_id:
            data["snapshot_id"] = snapshot_id
        if app_id is not None:
            data["app_id"] = app_id
        if image_id:
            data["image_id"] = image_id
        if ipxe_chain_url:
            data["ipxe_chain_url"] = ipxe_chain_url
        if script_id:
            data["script_id"] = script_id
        if enable_ipv6:
            data["enable_ipv6"] = enable_ipv6
        if disable_public_ipv4:
            data["disable_public_ipv4"] = disable_public_ipv4
        if attach_private_network:
            data["attach_private_network"] = attach_private_network
        if attach_vpc:
            data["attach_vpc"] = attach_vpc
        if attach_vpc2:
            data["attach_vpc2"] = attach_vpc2
        if label:
            data["label"] = label
        if sshkey_id:
            data["sshkey_id"] = sshkey_id
        if backups:
            data["backups"] = backups
        if ddos_protection:
            data["ddos_protection"] = ddos_protection
        if activation_email:
            data["activation_email"] = activation_email
        if hostname:
            data["hostname"] = hostname
        if tag:
            data["tag"] = tag
        if tags:
            data["tags"] = tags
        if firewall_group_id:
            data["firewall_group_id"] = firewall_group_id
        if reserved_ipv4:
            data["reserved_ipv4"] = reserved_ipv4
        if user_data:
            data["user_data"] = user_data

        response = self.client.post("instances", data)
        return response.get("instance", {})

    def update(
        self,
        instance_id: str,
        app_id: int = None,
        image_id: str = None,
        backups: str = None,
        firewall_group_id: str = None,
        enable_ipv6: bool = None,
        os_id: int = None,
        user_data: str = None,
        tag: str = None,
        tags: List[str] = None,
        label: str = None,
        ddos_protection: bool = None,
    ) -> Dict:
        """
        Update an instance

        Args:
            instance_id: Instance ID to update
            (other args same as create)

        Returns:
            Updated instance details
        """
        data = {}

        if app_id is not None:
            data["app_id"] = app_id
        if image_id is not None:
            data["image_id"] = image_id
        if backups is not None:
            data["backups"] = backups
        if firewall_group_id is not None:
            data["firewall_group_id"] = firewall_group_id
        if enable_ipv6 is not None:
            data["enable_ipv6"] = enable_ipv6
        if os_id is not None:
            data["os_id"] = os_id
        if user_data is not None:
            data["user_data"] = user_data
        if tag is not None:
            data["tag"] = tag
        if tags is not None:
            data["tags"] = tags
        if label is not None:
            data["label"] = label
        if ddos_protection is not None:
            data["ddos_protection"] = ddos_protection

        response = self.client.patch(f"instances/{instance_id}", data)
        return response.get("instance", {})

    def delete(self, instance_id: str) -> None:
        """
        Delete an instance

        Args:
            instance_id: Instance ID to delete
        """
        self.client.delete(f"instances/{instance_id}")

    def start(self, instance_id: str) -> None:
        """Start an instance"""
        self.client.post(f"instances/{instance_id}/start")

    def stop(self, instance_id: str) -> None:
        """Stop an instance (alias for halt)"""
        self.halt(instance_id)

    def halt(self, instance_id: str) -> None:
        """Halt an instance"""
        self.client.post(f"instances/{instance_id}/halt")

    def reboot(self, instance_id: str) -> None:
        """Reboot an instance"""
        self.client.post(f"instances/{instance_id}/reboot")

    def reinstall(self, instance_id: str, hostname: str = None) -> Dict:
        """
        Reinstall an instance

        Args:
            instance_id: Instance ID
            hostname: New hostname (optional)

        Returns:
            Instance details
        """
        data = {}
        if hostname:
            data["hostname"] = hostname

        response = self.client.post(f"instances/{instance_id}/reinstall", data)
        return response.get("instance", {})

    def get_bandwidth(self, instance_id: str) -> Dict:
        """Get instance bandwidth usage"""
        return self.client.get(f"instances/{instance_id}/bandwidth")

    def get_neighbors(self, instance_id: str) -> Dict:
        """Get instance neighbors (shared host)"""
        return self.client.get(f"instances/{instance_id}/neighbors")

    def get_user_data(self, instance_id: str) -> Dict:
        """Get instance user data"""
        return self.client.get(f"instances/{instance_id}/user-data")

    def get_upgrades(self, instance_id: str, upgrade_type: str = None) -> Dict:
        """
        Get available upgrades

        Args:
            instance_id: Instance ID
            upgrade_type: Filter by type ("all", "applications", "os", "plans")

        Returns:
            Available upgrades
        """
        params = {}
        if upgrade_type:
            params["type"] = upgrade_type
        return self.client.get(f"instances/{instance_id}/upgrades", params=params)

    # IPv4 management
    def list_ipv4(self, instance_id: str) -> List[Dict]:
        """List IPv4 addresses"""
        response = self.client.get(f"instances/{instance_id}/ipv4")
        return response.get("ipv4s", [])

    def create_ipv4(self, instance_id: str, reboot: bool = True) -> Dict:
        """
        Create additional IPv4

        Args:
            instance_id: Instance ID
            reboot: Reboot instance to apply (default True)

        Returns:
            New IPv4 info
        """
        response = self.client.post(
            f"instances/{instance_id}/ipv4",
            {"reboot": reboot}
        )
        return response.get("ipv4", {})

    def delete_ipv4(self, instance_id: str, ipv4: str) -> None:
        """Delete an IPv4 address"""
        self.client.delete(f"instances/{instance_id}/ipv4/{ipv4}")

    def create_ipv4_reverse(
        self,
        instance_id: str,
        ip: str,
        reverse: str
    ) -> None:
        """Set IPv4 reverse DNS"""
        self.client.post(
            f"instances/{instance_id}/ipv4/reverse",
            {"ip": ip, "reverse": reverse}
        )

    def set_default_ipv4_reverse(self, instance_id: str, ip: str) -> None:
        """Set default reverse DNS for IPv4"""
        self.client.post(
            f"instances/{instance_id}/ipv4/reverse/default",
            {"ip": ip}
        )

    # IPv6 management
    def list_ipv6(self, instance_id: str) -> List[Dict]:
        """List IPv6 addresses"""
        response = self.client.get(f"instances/{instance_id}/ipv6")
        return response.get("ipv6s", [])

    def create_ipv6_reverse(
        self,
        instance_id: str,
        ip: str,
        reverse: str
    ) -> None:
        """Set IPv6 reverse DNS"""
        self.client.post(
            f"instances/{instance_id}/ipv6/reverse",
            {"ip": ip, "reverse": reverse}
        )

    def list_ipv6_reverse(self, instance_id: str) -> List[Dict]:
        """List IPv6 reverse DNS entries"""
        response = self.client.get(f"instances/{instance_id}/ipv6/reverse")
        return response.get("reverse_ipv6s", [])

    def delete_ipv6_reverse(self, instance_id: str, ipv6: str) -> None:
        """Delete IPv6 reverse DNS"""
        self.client.delete(f"instances/{instance_id}/ipv6/reverse/{ipv6}")

    # Backup management
    def get_backup_schedule(self, instance_id: str) -> Dict:
        """Get backup schedule"""
        return self.client.get(f"instances/{instance_id}/backup-schedule")

    def set_backup_schedule(
        self,
        instance_id: str,
        backup_type: str,
        hour: int = None,
        dow: int = None,
        dom: int = None,
    ) -> None:
        """
        Set backup schedule

        Args:
            instance_id: Instance ID
            backup_type: Schedule type ("daily", "weekly", "monthly", "daily_alt_even", "daily_alt_odd")
            hour: Hour of day (0-23)
            dow: Day of week (1-7, for weekly)
            dom: Day of month (1-28, for monthly)
        """
        data = {"type": backup_type}
        if hour is not None:
            data["hour"] = hour
        if dow is not None:
            data["dow"] = dow
        if dom is not None:
            data["dom"] = dom

        self.client.post(f"instances/{instance_id}/backup-schedule", data)

    def restore_backup(self, instance_id: str, backup_id: str) -> Dict:
        """
        Restore instance from backup

        Args:
            instance_id: Instance ID
            backup_id: Backup ID to restore

        Returns:
            Restore status
        """
        return self.client.post(
            f"instances/{instance_id}/restore",
            {"backup_id": backup_id}
        )

    def restore_snapshot(self, instance_id: str, snapshot_id: str) -> Dict:
        """
        Restore instance from snapshot

        Args:
            instance_id: Instance ID
            snapshot_id: Snapshot ID to restore

        Returns:
            Restore status
        """
        return self.client.post(
            f"instances/{instance_id}/restore",
            {"snapshot_id": snapshot_id}
        )

    # VPC management
    def list_vpcs(self, instance_id: str) -> List[Dict]:
        """List VPCs attached to instance"""
        response = self.client.get(f"instances/{instance_id}/vpcs")
        return response.get("vpcs", [])

    def attach_vpc(self, instance_id: str, vpc_id: str) -> None:
        """Attach VPC to instance"""
        self.client.post(
            f"instances/{instance_id}/vpcs/attach",
            {"vpc_id": vpc_id}
        )

    def detach_vpc(self, instance_id: str, vpc_id: str) -> None:
        """Detach VPC from instance"""
        self.client.post(
            f"instances/{instance_id}/vpcs/detach",
            {"vpc_id": vpc_id}
        )

    # VPC 2.0 management
    def list_vpc2(self, instance_id: str) -> List[Dict]:
        """List VPC 2.0 networks attached to instance"""
        response = self.client.get(f"instances/{instance_id}/vpc2")
        return response.get("vpcs", [])

    def attach_vpc2(
        self,
        instance_id: str,
        vpc_id: str,
        ip_address: str = None
    ) -> None:
        """
        Attach VPC 2.0 to instance

        Args:
            instance_id: Instance ID
            vpc_id: VPC 2.0 ID
            ip_address: Specific IP to assign (optional)
        """
        data = {"vpc_id": vpc_id}
        if ip_address:
            data["ip_address"] = ip_address

        self.client.post(f"instances/{instance_id}/vpc2/attach", data)

    def detach_vpc2(self, instance_id: str, vpc_id: str) -> None:
        """Detach VPC 2.0 from instance"""
        self.client.post(
            f"instances/{instance_id}/vpc2/detach",
            {"vpc_id": vpc_id}
        )

    # ISO management
    def attach_iso(self, instance_id: str, iso_id: str) -> Dict:
        """Attach ISO to instance"""
        return self.client.post(
            f"instances/{instance_id}/iso/attach",
            {"iso_id": iso_id}
        )

    def detach_iso(self, instance_id: str) -> Dict:
        """Detach ISO from instance"""
        return self.client.post(f"instances/{instance_id}/iso/detach")

    def get_iso_status(self, instance_id: str) -> Dict:
        """Get ISO attach status"""
        return self.client.get(f"instances/{instance_id}/iso")