vultr-api/vultr_api/resources/load_balancers.py

"""
Load Balancers Resource

Load balancer management
"""

from typing import Dict, List
from .base import BaseResource


class LoadBalancersResource(BaseResource):
    """
    Load balancer management

    Usage:
        # List load balancers
        lbs = client.load_balancers.list()

        # Create load balancer
        lb = client.load_balancers.create(
            region="ewr",
            label="my-lb",
            forwarding_rules=[{
                "frontend_protocol": "http",
                "frontend_port": 80,
                "backend_protocol": "http",
                "backend_port": 80
            }]
        )
    """

    def list(self, per_page: int = 100, cursor: str = None) -> Dict:
        """
        List load balancers

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

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

    def get(self, load_balancer_id: str) -> Dict:
        """
        Get load balancer details

        Args:
            load_balancer_id: Load balancer ID

        Returns:
            Load balancer details
        """
        response = self.client.get(f"load-balancers/{load_balancer_id}")
        return response.get("load_balancer", {})

    def create(
        self,
        region: str,
        forwarding_rules: List[Dict],
        label: str = None,
        balancing_algorithm: str = "roundrobin",
        ssl_redirect: bool = False,
        proxy_protocol: bool = False,
        health_check: Dict = None,
        sticky_session: Dict = None,
        ssl: Dict = None,
        instances: List[str] = None,
        firewall_rules: List[Dict] = None,
        vpc: str = None,
    ) -> Dict:
        """
        Create a load balancer

        Args:
            region: Region ID
            forwarding_rules: List of forwarding rules
            label: Load balancer label
            balancing_algorithm: "roundrobin" or "leastconn"
            ssl_redirect: Redirect HTTP to HTTPS
            proxy_protocol: Enable proxy protocol
            health_check: Health check config
            sticky_session: Sticky session config
            ssl: SSL config
            instances: List of instance IDs to attach
            firewall_rules: List of firewall rules
            vpc: VPC ID

        Returns:
            Created load balancer details

        Example forwarding_rules:
            [{
                "frontend_protocol": "http",
                "frontend_port": 80,
                "backend_protocol": "http",
                "backend_port": 80
            }]

        Example health_check:
            {
                "protocol": "http",
                "port": 80,
                "path": "/health",
                "check_interval": 15,
                "response_timeout": 5,
                "unhealthy_threshold": 5,
                "healthy_threshold": 5
            }
        """
        data = {
            "region": region,
            "forwarding_rules": forwarding_rules,
            "balancing_algorithm": balancing_algorithm,
            "ssl_redirect": ssl_redirect,
            "proxy_protocol": proxy_protocol,
        }

        if label:
            data["label"] = label
        if health_check:
            data["health_check"] = health_check
        if sticky_session:
            data["sticky_session"] = sticky_session
        if ssl:
            data["ssl"] = ssl
        if instances:
            data["instances"] = instances
        if firewall_rules:
            data["firewall_rules"] = firewall_rules
        if vpc:
            data["vpc"] = vpc

        response = self.client.post("load-balancers", data)
        return response.get("load_balancer", {})

    def update(
        self,
        load_balancer_id: str,
        label: str = None,
        balancing_algorithm: str = None,
        ssl_redirect: bool = None,
        proxy_protocol: bool = None,
        health_check: Dict = None,
        forwarding_rules: List[Dict] = None,
        sticky_session: Dict = None,
        ssl: Dict = None,
        instances: List[str] = None,
        firewall_rules: List[Dict] = None,
        vpc: str = None,
    ) -> None:
        """
        Update a load balancer

        Args:
            load_balancer_id: Load balancer ID
            (other args same as create)
        """
        data = {}

        if label is not None:
            data["label"] = label
        if balancing_algorithm is not None:
            data["balancing_algorithm"] = balancing_algorithm
        if ssl_redirect is not None:
            data["ssl_redirect"] = ssl_redirect
        if proxy_protocol is not None:
            data["proxy_protocol"] = proxy_protocol
        if health_check is not None:
            data["health_check"] = health_check
        if forwarding_rules is not None:
            data["forwarding_rules"] = forwarding_rules
        if sticky_session is not None:
            data["sticky_session"] = sticky_session
        if ssl is not None:
            data["ssl"] = ssl
        if instances is not None:
            data["instances"] = instances
        if firewall_rules is not None:
            data["firewall_rules"] = firewall_rules
        if vpc is not None:
            data["vpc"] = vpc

        self.client.patch(f"load-balancers/{load_balancer_id}", data)

    def delete(self, load_balancer_id: str) -> None:
        """
        Delete a load balancer

        Args:
            load_balancer_id: Load balancer ID to delete
        """
        self.client.delete(f"load-balancers/{load_balancer_id}")

    # Forwarding rules
    def list_forwarding_rules(self, load_balancer_id: str) -> List[Dict]:
        """List forwarding rules"""
        response = self.client.get(
            f"load-balancers/{load_balancer_id}/forwarding-rules"
        )
        return response.get("forwarding_rules", [])

    def create_forwarding_rule(
        self,
        load_balancer_id: str,
        frontend_protocol: str,
        frontend_port: int,
        backend_protocol: str,
        backend_port: int
    ) -> Dict:
        """Create a forwarding rule"""
        data = {
            "frontend_protocol": frontend_protocol,
            "frontend_port": frontend_port,
            "backend_protocol": backend_protocol,
            "backend_port": backend_port,
        }
        response = self.client.post(
            f"load-balancers/{load_balancer_id}/forwarding-rules",
            data
        )
        return response.get("forwarding_rule", {})

    def get_forwarding_rule(
        self,
        load_balancer_id: str,
        rule_id: str
    ) -> Dict:
        """Get a forwarding rule"""
        response = self.client.get(
            f"load-balancers/{load_balancer_id}/forwarding-rules/{rule_id}"
        )
        return response.get("forwarding_rule", {})

    def delete_forwarding_rule(
        self,
        load_balancer_id: str,
        rule_id: str
    ) -> None:
        """Delete a forwarding rule"""
        self.client.delete(
            f"load-balancers/{load_balancer_id}/forwarding-rules/{rule_id}"
        )

    # Firewall rules
    def list_firewall_rules(self, load_balancer_id: str) -> List[Dict]:
        """List firewall rules"""
        response = self.client.get(
            f"load-balancers/{load_balancer_id}/firewall-rules"
        )
        return response.get("firewall_rules", [])

    def get_firewall_rule(
        self,
        load_balancer_id: str,
        rule_id: str
    ) -> Dict:
        """Get a firewall rule"""
        response = self.client.get(
            f"load-balancers/{load_balancer_id}/firewall-rules/{rule_id}"
        )
        return response.get("firewall_rule", {})