From 87c2bd00fb3c6cb7f8a74bbad9a95cb4f7daf1eb Mon Sep 17 00:00:00 2001 From: AseemPrasad Date: Tue, 7 Oct 2025 22:47:37 +0530 Subject: [PATCH] fix:updated poc_droplets_volumes_sshkeys.py --- examples/poc_droplets_volumes_sshkeys.py | 33 ++++++++++++------------ 1 file changed, 17 insertions(+), 16 deletions(-) diff --git a/examples/poc_droplets_volumes_sshkeys.py b/examples/poc_droplets_volumes_sshkeys.py index e313de7f..81dd9c02 100644 --- a/examples/poc_droplets_volumes_sshkeys.py +++ b/examples/poc_droplets_volumes_sshkeys.py @@ -18,17 +18,17 @@ class DigitalOceanError(Exception): class DropletCreator: def __init__(self, *args, **kwargs): token = os.environ.get("DIGITALOCEAN_TOKEN") - if token == "": - raise Exception("No DigitalOcean API token in DIGITALOCEAN_TOKEN env var") - self.client = Client(token=os.environ.get("DIGITALOCEAN_TOKEN")) + if not token: + raise DigitalOceanError("No DigitalOcean API token in DIGITALOCEAN_TOKEN env var") + self.client = Client(token=token) def throw(self, message): raise DigitalOceanError(message) from None def main(self): key_name = os.environ.get("SSH_KEY_NAME") - if key_name == "": - raise Exception("SSH_KEY_NAME not set") + if not key_name: + raise DigitalOceanError("SSH_KEY_NAME not set") ssh_key = self.find_ssh_key(key_name) droplet_req = { @@ -67,7 +67,9 @@ def main(self): print("Done!") - def create_droplet(self, req={}): + def create_droplet(self, req=None): + if req is None: + req = {} print("Creating Droplet using: {0}".format(req)) try: resp = self.client.droplets.create(body=req) @@ -77,9 +79,6 @@ def create_droplet(self, req={}): get_resp = self.client.droplets.get(droplet_id) droplet = get_resp["droplet"] ip_address = "" - # Would be nice if we could surface the IP address somehow. - # For example godo has the PublicIPv4 method: - # https://github.com/digitalocean/godo/blob/a084002940af6a9b818e3c8fb31a4920356fbb75/droplets.go#L66-L79 for net in droplet["networks"]["v4"]: if net["type"] == "public": ip_address = net["ip_address"] @@ -115,7 +114,7 @@ def wait_for_action(self, id, wait=5): print(".", end="", flush=True) sleep(wait) elif status == "errored": - raise Exception( + raise DigitalOceanError( "{0} action {1} {2}".format( resp["action"]["type"], resp["action"]["id"], status ) @@ -141,17 +140,19 @@ def find_ssh_key(self, name): ) ) - pages = resp.links.pages - if "next" in pages.keys(): - # Having to parse the URL to find the next page is not very friendly. + # Fix: Use dict access for links/pages and int for page + pages = resp.get("links", {}).get("pages", {}) + if "next" in pages: parsed_url = urlparse(pages["next"]) - page = parse_qs(parsed_url.query)["page"][0] + page = int(parse_qs(parsed_url.query)["page"][0]) else: paginated = False - raise Exception("no ssh key found") + raise DigitalOceanError("no ssh key found") - def create_volume(self, req={}): + def create_volume(self, req=None): + if req is None: + req = {} print("Creating volume using: {0}".format(req)) try: resp = self.client.volumes.create(body=req)