""" 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")