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>
136 lines
3.4 KiB
Python
136 lines
3.4 KiB
Python
"""
|
|
Block Storage Resource
|
|
|
|
Block storage management
|
|
"""
|
|
|
|
from typing import Dict, List
|
|
from .base import BaseResource
|
|
|
|
|
|
class BlockStorageResource(BaseResource):
|
|
"""
|
|
Block storage management
|
|
|
|
Usage:
|
|
# List block storage
|
|
blocks = client.block_storage.list()
|
|
|
|
# Create block storage
|
|
block = client.block_storage.create(
|
|
region="ewr",
|
|
size_gb=50,
|
|
label="my-storage"
|
|
)
|
|
|
|
# Attach to instance
|
|
client.block_storage.attach(block_id="block-id", instance_id="instance-id")
|
|
"""
|
|
|
|
def list(self, per_page: int = 100, cursor: str = None) -> Dict:
|
|
"""
|
|
List block storage volumes
|
|
|
|
Returns:
|
|
Dict with 'blocks' list and 'meta' pagination
|
|
"""
|
|
params = {"per_page": per_page}
|
|
if cursor:
|
|
params["cursor"] = cursor
|
|
return self.client.get("blocks", params=params)
|
|
|
|
def list_all(self) -> List[Dict]:
|
|
"""List all block storage volumes (auto-paginate)"""
|
|
return self.client.paginate("blocks", "blocks")
|
|
|
|
def get(self, block_id: str) -> Dict:
|
|
"""
|
|
Get block storage details
|
|
|
|
Args:
|
|
block_id: Block storage ID
|
|
|
|
Returns:
|
|
Block storage details
|
|
"""
|
|
response = self.client.get(f"blocks/{block_id}")
|
|
return response.get("block", {})
|
|
|
|
def create(
|
|
self,
|
|
region: str,
|
|
size_gb: int,
|
|
label: str = None,
|
|
block_type: str = None
|
|
) -> Dict:
|
|
"""
|
|
Create block storage
|
|
|
|
Args:
|
|
region: Region ID
|
|
size_gb: Size in GB (10-40000)
|
|
label: Label
|
|
block_type: Storage type ("high_perf" or "storage_opt")
|
|
|
|
Returns:
|
|
Created block storage details
|
|
"""
|
|
data = {"region": region, "size_gb": size_gb}
|
|
if label:
|
|
data["label"] = label
|
|
if block_type:
|
|
data["block_type"] = block_type
|
|
|
|
response = self.client.post("blocks", data)
|
|
return response.get("block", {})
|
|
|
|
def update(self, block_id: str, label: str = None, size_gb: int = None) -> None:
|
|
"""
|
|
Update block storage
|
|
|
|
Args:
|
|
block_id: Block ID
|
|
label: New label
|
|
size_gb: New size (can only increase)
|
|
"""
|
|
data = {}
|
|
if label:
|
|
data["label"] = label
|
|
if size_gb:
|
|
data["size_gb"] = size_gb
|
|
|
|
self.client.patch(f"blocks/{block_id}", data)
|
|
|
|
def delete(self, block_id: str) -> None:
|
|
"""
|
|
Delete block storage
|
|
|
|
Args:
|
|
block_id: Block ID to delete
|
|
"""
|
|
self.client.delete(f"blocks/{block_id}")
|
|
|
|
def attach(self, block_id: str, instance_id: str, live: bool = False) -> None:
|
|
"""
|
|
Attach block storage to instance
|
|
|
|
Args:
|
|
block_id: Block storage ID
|
|
instance_id: Instance ID to attach to
|
|
live: Live attach (no reboot)
|
|
"""
|
|
self.client.post(f"blocks/{block_id}/attach", {
|
|
"instance_id": instance_id,
|
|
"live": live
|
|
})
|
|
|
|
def detach(self, block_id: str, live: bool = False) -> None:
|
|
"""
|
|
Detach block storage from instance
|
|
|
|
Args:
|
|
block_id: Block storage ID
|
|
live: Live detach (no reboot)
|
|
"""
|
|
self.client.post(f"blocks/{block_id}/detach", {"live": live})
|