147 lines
4.6 KiB
Python
147 lines
4.6 KiB
Python
import argparse
|
|
import keyring
|
|
import requests
|
|
|
|
|
|
class Scaleway:
|
|
API_ENDPOINT_BASE = "https://api.scaleway.com/instance/v1/zones"
|
|
ZONES = [
|
|
"fr-par-1", "fr-par-2", "fr-par-3",
|
|
"nl-ams-1", "nl-ams-2",
|
|
"pl-waw-1", "pl-waw-2",
|
|
]
|
|
|
|
def __init__(self, project_id, secret_key):
|
|
self.__zone = None
|
|
self.__project_id = project_id
|
|
self.__headers = {"X-Auth-Token": secret_key}
|
|
|
|
@property
|
|
def zone(self):
|
|
return self.__zone
|
|
|
|
@zone.setter
|
|
def zone(self, zone):
|
|
if zone not in Scaleway.ZONES:
|
|
raise KeyError(f"{zone} is not a valid zone - must be one of {Scaleway.ZONES}")
|
|
self.__zone = zone
|
|
|
|
@property
|
|
def project_id(self):
|
|
return self.__project_id
|
|
|
|
def __url(self, item, id):
|
|
if self.__zone is None:
|
|
raise RuntimeError("zone must be set before making any API requests")
|
|
|
|
url = f"{Scaleway.API_ENDPOINT_BASE}/{self.__zone}"
|
|
|
|
if id == "products":
|
|
return f"{url}/products/{item}"
|
|
|
|
url = f"{url}/{item}"
|
|
if id is not None:
|
|
url = f"{url}/{id}"
|
|
|
|
return url
|
|
|
|
@staticmethod
|
|
def __check_status(type, url, rsp):
|
|
if (rsp.status_code // 100) != 2:
|
|
raise RuntimeError(
|
|
f"{type} {url} returned with status code {rsp.status_code}: {rsp.json()}")
|
|
|
|
def get(self, item, id=None):
|
|
url = self.__url(item, id)
|
|
r = requests.get(url, headers=self.__headers)
|
|
self.__check_status("GET", url, r)
|
|
return r.json()[item]
|
|
|
|
def get_by_name(self, item, name):
|
|
items = self.get(item)
|
|
return next((it for it in items if it["name"] == name), None)
|
|
|
|
def __post(self, url, data):
|
|
r = requests.post(url, headers=self.__headers, json=data)
|
|
self.__check_status("POST", url, r)
|
|
return r.json()
|
|
|
|
def post(self, item, data):
|
|
return self.__post(self.__url(item, None), data)
|
|
|
|
def post_action(self, item, id, action, data):
|
|
return self.__post(f"{self.__url(item, id)}/{action}", data)
|
|
|
|
def delete(self, item, id):
|
|
url = self.__url(item, id)
|
|
r = requests.delete(url, headers=self.__headers)
|
|
self.__check_status("DELETE", url, r)
|
|
|
|
|
|
def create_baldur(scaleway, args):
|
|
volume_size = args.volume_size
|
|
|
|
security_group = scaleway.get_by_name("security_groups", "baldur-security-group")
|
|
image = scaleway.get_by_name("images", "Debian Bullseye")
|
|
server_type = "PLAY2-PICO"
|
|
if server_type not in scaleway.get("servers", id="products"):
|
|
raise RuntimeError(f"{server_type} is not available in {scaleway.zone}")
|
|
|
|
response = scaleway.post("ips", data={"project": scaleway.project_id})
|
|
public_ip = response["ip"]
|
|
|
|
baldur = {
|
|
"name": "baldur",
|
|
"dynamic_ip_required": False,
|
|
"commercial_type": server_type,
|
|
"image": image["id"],
|
|
"volumes": {"0": {"size": int(volume_size * 1_000_000_000)}},
|
|
"enable_ipv6": False,
|
|
"public_ip": public_ip["id"],
|
|
"project": scaleway.project_id,
|
|
"security_group": security_group["id"],
|
|
}
|
|
|
|
response = scaleway.post("servers", data=baldur)
|
|
server = response["server"]
|
|
|
|
scaleway.post_action("servers", server["id"], "action", data={"action": "poweron"})
|
|
|
|
print("Baldur instance created:")
|
|
print(f" block volume size: {server['volumes']['0']['size']//1_000_000_000} GiB")
|
|
print(f" public ip address: {server['public_ip']['address']}")
|
|
|
|
|
|
def delete_baldur(scaleway, _):
|
|
server = scaleway.get_by_name("servers", "baldur")
|
|
if server is None:
|
|
raise RuntimeError(f"Baldur instance was not found in {scaleway.zone}")
|
|
ip = server["public_ip"]
|
|
|
|
scaleway.post_action("servers", server["id"], "action", data={"action": "terminate"})
|
|
scaleway.delete("ips", ip["id"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Create or delete the Baldur instance")
|
|
|
|
subparsers = parser.add_subparsers()
|
|
|
|
create_parser = subparsers.add_parser("create")
|
|
create_parser.add_argument("--volume-size", type=int, required=True,
|
|
help="Block volume size (in GiB) to create")
|
|
create_parser.set_defaults(func=create_baldur)
|
|
|
|
delete_parser = subparsers.add_parser("delete")
|
|
delete_parser.set_defaults(func=delete_baldur)
|
|
|
|
args = parser.parse_args()
|
|
|
|
scw_project_id = keyring.get_password("scaleway", "project_id")
|
|
scw_secret_key = keyring.get_password("scaleway", "secret_key")
|
|
|
|
scaleway = Scaleway(scw_project_id, scw_secret_key)
|
|
scaleway.zone = "fr-par-2"
|
|
|
|
args.func(scaleway, args)
|