ansible-edda/scripts/scaleway/baldur.py

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)