vultr-api/vultr_api/resources/vpc.py

"""
VPC Resource

VPC (Virtual Private Cloud) management - both v1 and v2
"""

from typing import Dict, List
from .base import BaseResource


class VPCResource(BaseResource):
    """
    VPC management (v1 and v2)

    Usage:
        # List VPCs
        vpcs = client.vpc.list()

        # Create VPC
        vpc = client.vpc.create(
            region="ewr",
            description="My VPC",
            v4_subnet="10.0.0.0",
            v4_subnet_mask=24
        )

        # VPC 2.0
        vpc2 = client.vpc.create_vpc2(region="ewr", description="My VPC 2.0")
    """

    # VPC v1
    def list(self, per_page: int = 100, cursor: str = None) -> Dict:
        """
        List VPCs

        Returns:
            Dict with 'vpcs' list and 'meta' pagination
        """
        params = {"per_page": per_page}
        if cursor:
            params["cursor"] = cursor
        return self.client.get("vpcs", params=params)

    def list_all(self) -> List[Dict]:
        """List all VPCs (auto-paginate)"""
        return self.client.paginate("vpcs", "vpcs")

    def get(self, vpc_id: str) -> Dict:
        """
        Get VPC details

        Args:
            vpc_id: VPC ID

        Returns:
            VPC details
        """
        response = self.client.get(f"vpcs/{vpc_id}")
        return response.get("vpc", {})

    def create(
        self,
        region: str,
        description: str = None,
        v4_subnet: str = None,
        v4_subnet_mask: int = None
    ) -> Dict:
        """
        Create a VPC

        Args:
            region: Region ID
            description: VPC description
            v4_subnet: IPv4 subnet (e.g., "10.0.0.0")
            v4_subnet_mask: Subnet mask (e.g., 24)

        Returns:
            Created VPC details
        """
        data = {"region": region}
        if description:
            data["description"] = description
        if v4_subnet:
            data["v4_subnet"] = v4_subnet
        if v4_subnet_mask:
            data["v4_subnet_mask"] = v4_subnet_mask

        response = self.client.post("vpcs", data)
        return response.get("vpc", {})

    def update(self, vpc_id: str, description: str) -> None:
        """
        Update VPC description

        Args:
            vpc_id: VPC ID
            description: New description
        """
        self.client.put(f"vpcs/{vpc_id}", {"description": description})

    def delete(self, vpc_id: str) -> None:
        """
        Delete a VPC

        Args:
            vpc_id: VPC ID to delete
        """
        self.client.delete(f"vpcs/{vpc_id}")

    # VPC 2.0
    def list_vpc2(self, per_page: int = 100, cursor: str = None) -> Dict:
        """
        List VPC 2.0 networks

        Returns:
            Dict with 'vpcs' list and 'meta' pagination
        """
        params = {"per_page": per_page}
        if cursor:
            params["cursor"] = cursor
        return self.client.get("vpc2", params=params)

    def list_all_vpc2(self) -> List[Dict]:
        """List all VPC 2.0 networks (auto-paginate)"""
        return self.client.paginate("vpc2", "vpcs")

    def get_vpc2(self, vpc_id: str) -> Dict:
        """
        Get VPC 2.0 details

        Args:
            vpc_id: VPC 2.0 ID

        Returns:
            VPC 2.0 details
        """
        response = self.client.get(f"vpc2/{vpc_id}")
        return response.get("vpc", {})

    def create_vpc2(
        self,
        region: str,
        description: str = None,
        ip_block: str = None,
        prefix_length: int = None
    ) -> Dict:
        """
        Create a VPC 2.0 network

        Args:
            region: Region ID
            description: VPC description
            ip_block: IP block (e.g., "10.0.0.0")
            prefix_length: Prefix length (e.g., 24)

        Returns:
            Created VPC 2.0 details
        """
        data = {"region": region}
        if description:
            data["description"] = description
        if ip_block:
            data["ip_block"] = ip_block
        if prefix_length:
            data["prefix_length"] = prefix_length

        response = self.client.post("vpc2", data)
        return response.get("vpc", {})

    def update_vpc2(self, vpc_id: str, description: str) -> None:
        """
        Update VPC 2.0 description

        Args:
            vpc_id: VPC 2.0 ID
            description: New description
        """
        self.client.put(f"vpc2/{vpc_id}", {"description": description})

    def delete_vpc2(self, vpc_id: str) -> None:
        """
        Delete a VPC 2.0 network

        Args:
            vpc_id: VPC 2.0 ID to delete
        """
        self.client.delete(f"vpc2/{vpc_id}")

    def list_vpc2_nodes(self, vpc_id: str) -> List[Dict]:
        """
        List nodes attached to VPC 2.0

        Args:
            vpc_id: VPC 2.0 ID

        Returns:
            List of attached nodes
        """
        response = self.client.get(f"vpc2/{vpc_id}/nodes")
        return response.get("nodes", [])

    def attach_vpc2_nodes(self, vpc_id: str, nodes: List[Dict]) -> None:
        """
        Attach nodes to VPC 2.0

        Args:
            vpc_id: VPC 2.0 ID
            nodes: List of nodes [{"id": "instance-id", "ip_address": "10.0.0.5"}, ...]
        """
        self.client.post(f"vpc2/{vpc_id}/nodes/attach", {"nodes": nodes})

    def detach_vpc2_nodes(self, vpc_id: str, nodes: List[str]) -> None:
        """
        Detach nodes from VPC 2.0

        Args:
            vpc_id: VPC 2.0 ID
            nodes: List of node IDs
        """
        self.client.post(f"vpc2/{vpc_id}/nodes/detach", {"nodes": nodes})