184054c6c1
- vultr_api/: Python library wrapping Vultr API v2 - 17 resource modules (instances, dns, firewall, vpc, etc.) - Pagination support, error handling - server/: FastAPI REST server - All API endpoints exposed via HTTP - X-API-Key header authentication - Swagger docs at /docs - Podman quadlet config for systemd deployment Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""Load Balancers 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 ForwardingRule(BaseModel):
|
|
frontend_protocol: str
|
|
frontend_port: int
|
|
backend_protocol: str
|
|
backend_port: int
|
|
|
|
|
|
class HealthCheck(BaseModel):
|
|
protocol: str
|
|
port: int
|
|
path: Optional[str] = None
|
|
check_interval: Optional[int] = None
|
|
response_timeout: Optional[int] = None
|
|
unhealthy_threshold: Optional[int] = None
|
|
healthy_threshold: Optional[int] = None
|
|
|
|
|
|
class CreateLoadBalancerRequest(BaseModel):
|
|
region: str
|
|
label: Optional[str] = None
|
|
balancing_algorithm: Optional[str] = None # roundrobin or leastconn
|
|
forwarding_rules: Optional[List[ForwardingRule]] = None
|
|
health_check: Optional[HealthCheck] = None
|
|
instances: Optional[List[str]] = None
|
|
ssl_redirect: Optional[bool] = None
|
|
proxy_protocol: Optional[bool] = None
|
|
|
|
|
|
class UpdateLoadBalancerRequest(BaseModel):
|
|
label: Optional[str] = None
|
|
balancing_algorithm: Optional[str] = None
|
|
ssl_redirect: Optional[bool] = None
|
|
proxy_protocol: Optional[bool] = None
|
|
|
|
|
|
@router.get("")
|
|
async def list_load_balancers(
|
|
per_page: int = Query(25, le=500),
|
|
cursor: Optional[str] = None,
|
|
client: VultrClient = Depends(get_client)
|
|
):
|
|
"""List all load balancers"""
|
|
return client.load_balancers.list(per_page=per_page, cursor=cursor)
|
|
|
|
|
|
@router.get("/all")
|
|
async def list_all_load_balancers(client: VultrClient = Depends(get_client)):
|
|
"""List all load balancers (auto-paginated)"""
|
|
return {"load_balancers": client.load_balancers.list_all()}
|
|
|
|
|
|
@router.get("/{lb_id}")
|
|
async def get_load_balancer(lb_id: str, client: VultrClient = Depends(get_client)):
|
|
"""Get load balancer details"""
|
|
return client.load_balancers.get(lb_id)
|
|
|
|
|
|
@router.post("")
|
|
async def create_load_balancer(req: CreateLoadBalancerRequest, client: VultrClient = Depends(get_client)):
|
|
"""Create a load balancer"""
|
|
data = req.model_dump(exclude_none=True)
|
|
if 'forwarding_rules' in data:
|
|
data['forwarding_rules'] = [r for r in data['forwarding_rules']]
|
|
if 'health_check' in data:
|
|
data['health_check'] = dict(data['health_check'])
|
|
return client.load_balancers.create(**data)
|
|
|
|
|
|
@router.patch("/{lb_id}")
|
|
async def update_load_balancer(lb_id: str, req: UpdateLoadBalancerRequest, client: VultrClient = Depends(get_client)):
|
|
"""Update a load balancer"""
|
|
return client.load_balancers.update(lb_id, **req.model_dump(exclude_none=True))
|
|
|
|
|
|
@router.delete("/{lb_id}")
|
|
async def delete_load_balancer(lb_id: str, client: VultrClient = Depends(get_client)):
|
|
"""Delete a load balancer"""
|
|
return client.load_balancers.delete(lb_id)
|
|
|
|
|
|
# Forwarding rules
|
|
@router.get("/{lb_id}/forwarding-rules")
|
|
async def list_forwarding_rules(lb_id: str, client: VultrClient = Depends(get_client)):
|
|
"""List forwarding rules for a load balancer"""
|
|
return client.load_balancers.list_forwarding_rules(lb_id)
|
|
|
|
|
|
@router.post("/{lb_id}/forwarding-rules")
|
|
async def create_forwarding_rule(lb_id: str, req: ForwardingRule, client: VultrClient = Depends(get_client)):
|
|
"""Create a forwarding rule"""
|
|
return client.load_balancers.create_forwarding_rule(lb_id, **req.model_dump())
|
|
|
|
|
|
@router.delete("/{lb_id}/forwarding-rules/{rule_id}")
|
|
async def delete_forwarding_rule(lb_id: str, rule_id: str, client: VultrClient = Depends(get_client)):
|
|
"""Delete a forwarding rule"""
|
|
return client.load_balancers.delete_forwarding_rule(lb_id, rule_id)
|