mirror of
https://gitlab.com/veilid/veilid.git
synced 2025-04-25 18:09:34 -04:00
114 lines
4.8 KiB
Python
114 lines
4.8 KiB
Python
import aiohttp
|
|
import asyncio
|
|
import sys
|
|
|
|
# Define droplet configurations for different droplet types.
|
|
DROPLET_CONFIGS = {
|
|
"amd64-deb": {
|
|
"name": "build-server-amd64-deb-tmp",
|
|
"image": 179066895,
|
|
"size": "c2-16vcpu-32gb"
|
|
},
|
|
}
|
|
|
|
async def create_droplet(token: str, droplet_type: str) -> None:
|
|
config = DROPLET_CONFIGS.get(droplet_type)
|
|
if not config:
|
|
print(f"Droplet type '{droplet_type}' not recognized.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
create_url = "https://api.digitalocean.com/v2/droplets"
|
|
payload = {
|
|
"name": config["name"],
|
|
"region": "nyc1", # Changed default region to "ncy1"
|
|
"size": config["size"],
|
|
"image": config["image"],
|
|
"backups": False,
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(create_url, headers=headers, json=payload) as resp:
|
|
if resp.status not in (201, 202):
|
|
error_text = await resp.text()
|
|
print(f"Error creating droplet: {error_text}", file=sys.stderr)
|
|
sys.exit(error_text)
|
|
data = await resp.json()
|
|
droplet = data.get("droplet")
|
|
if not droplet:
|
|
print("No droplet information returned.", file=sys.stderr)
|
|
sys.exit("No droplet information returned.")
|
|
droplet_id = droplet.get("id")
|
|
print(f"Droplet creation initiated. Droplet ID: {droplet_id}")
|
|
|
|
# Poll for droplet status until it becomes "active"
|
|
status = droplet.get("status", "new")
|
|
droplet_url = f"https://api.digitalocean.com/v2/droplets/{droplet_id}"
|
|
while status != "active":
|
|
await asyncio.sleep(2)
|
|
async with session.get(droplet_url, headers=headers) as poll_resp:
|
|
if poll_resp.status != 200:
|
|
error_text = await poll_resp.text()
|
|
print(f"Error polling droplet status: {error_text}", file=sys.stderr)
|
|
sys.exit(error_text)
|
|
droplet_data = await poll_resp.json()
|
|
droplet = droplet_data.get("droplet")
|
|
if droplet:
|
|
status = droplet.get("status", status)
|
|
print(f"Droplet status: {status}")
|
|
else:
|
|
print("Droplet data missing in polling response", file=sys.stderr)
|
|
sys.exit("Droplet data missing in polling response")
|
|
|
|
print("Droplet is up and running.")
|
|
# Once active, send a final GET request to output the droplet's information.
|
|
async with session.get(droplet_url, headers=headers) as final_resp:
|
|
if final_resp.status != 200:
|
|
error_text = await final_resp.text()
|
|
print(f"Error retrieving droplet information: {error_text}", file=sys.stderr)
|
|
sys.exit(error_text)
|
|
final_data = await final_resp.json()
|
|
print("Droplet Information:")
|
|
print(final_data)
|
|
|
|
async def delete_droplet(token: str, droplet_type: str) -> None:
|
|
config = DROPLET_CONFIGS.get(droplet_type)
|
|
if not config:
|
|
print(f"Droplet type '{droplet_type}' not recognized.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
droplets_url = "https://api.digitalocean.com/v2/droplets"
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(droplets_url, headers=headers) as resp:
|
|
if resp.status != 200:
|
|
error_text = await resp.text()
|
|
print(f"Error retrieving droplets: {error_text}", file=sys.stderr)
|
|
sys.exit(error_text)
|
|
data = await resp.json()
|
|
droplets = data.get("droplets", [])
|
|
target_droplet = None
|
|
for droplet in droplets:
|
|
if droplet.get("name") == config["name"]:
|
|
target_droplet = droplet
|
|
break
|
|
if not target_droplet:
|
|
print(f"No droplet found with name '{config['name']}'.")
|
|
return
|
|
|
|
droplet_id = target_droplet.get("id")
|
|
delete_url = f"https://api.digitalocean.com/v2/droplets/{droplet_id}"
|
|
async with session.delete(delete_url, headers=headers) as delete_resp:
|
|
if delete_resp.status != 204:
|
|
error_text = await delete_resp.text()
|
|
print(f"Error deleting droplet: {error_text}", file=sys.stderr)
|
|
sys.exit(error_text)
|
|
print(f"Droplet '{config['name']}' deleted successfully.")
|