# -*- coding: utf-8 -*- import re from .Action import Action from .Image import Image from .Kernel import Kernel from .baseapi import BaseAPI, Error, GET, POST, DELETE from .SSHKey import SSHKey from .Volume import Volume class DropletError(Error): """Base exception class for this module""" pass class BadKernelObject(DropletError): pass class BadSSHKeyFormat(DropletError): pass class Droplet(BaseAPI): """Droplet management Attributes accepted at creation time: Args: name (str): name size_slug (str): droplet size image (str): image name to use to create droplet region (str): region ssh_keys (:obj:`str`, optional): list of ssh keys backups (bool): True if backups enabled ipv6 (bool): True if ipv6 enabled private_networking (bool): True if private networking enabled user_data (str): arbitrary data to pass to droplet volumes (:obj:`str`, optional): list of blockstorage volumes monitoring (bool): True if installing the DigitalOcean monitoring agent vpc_uuid (str, optional): ID of a VPC in which the Droplet will be created Attributes returned by API: * id (int): droplet id * memory (str): memory size * vcpus (int): number of vcpus * disk (int): disk size in GB * locked (bool): True if locked * created_at (str): creation date in format u'2014-11-06T10:42:09Z' * status (str): status, e.g. 'new', 'active', etc * networks (dict): details of connected networks * kernel (dict): details of kernel * backup_ids (:obj:`int`, optional): list of ids of backups of this droplet * snapshot_ids (:obj:`int`, optional): list of ids of snapshots of this droplet * action_ids (:obj:`int`, optional): list of ids of actions * features (:obj:`str`, optional): list of enabled features. e.g. [u'private_networking', u'virtio'] * image (dict): details of image used to create this droplet * ip_address (str): public ip addresses * private_ip_address (str): private ip address * ip_v6_address (:obj:`str`, optional): list of ipv6 addresses assigned * end_point (str): url of api endpoint used * volume_ids (:obj:`str`, optional): list of blockstorage volumes * vpc_uuid (str, optional): ID of the VPC that the Droplet is assigned to """ def __init__(self, *args, **kwargs): # Defining default values self.id = None self.name = None self.memory = None self.vcpus = None self.disk = None self.region = [] self.image = None self.size_slug = None self.locked = None self.created_at = None self.status = None self.networks = [] self.kernel = None self.backup_ids = [] self.snapshot_ids = [] self.action_ids = [] self.features = [] self.ip_address = None self.private_ip_address = None self.ip_v6_address = None self.ssh_keys = [] self.backups = None self.ipv6 = None self.private_networking = None self.user_data = None self.volumes = [] self.tags = [] self.monitoring = None self.vpc_uuid = None # This will load also the values passed super(Droplet, self).__init__(*args, **kwargs) @classmethod def get_object(cls, api_token, droplet_id): """Class method that will return a Droplet object by ID. Args: api_token (str): token droplet_id (int): droplet id """ droplet = cls(token=api_token, id=droplet_id) droplet.load() return droplet @classmethod def create_multiple(*args, **kwargs): api = BaseAPI(token=kwargs.get("token")) data = { "names": kwargs.get("names"), "size": kwargs.get("size_slug") or kwargs.get("size"), "image": kwargs.get("image"), "region": kwargs.get("region"), "backups": bool(kwargs.get("backups")), "ipv6": bool(kwargs.get("ipv6")), "private_networking": bool(kwargs.get("private_networking")), "tags": kwargs.get("tags"), "monitoring": bool(kwargs.get("monitoring")), "vpc_uuid": kwargs.get("vpc_uuid"), } if kwargs.get("ssh_keys"): data["ssh_keys"] = Droplet.__get_ssh_keys_id_or_fingerprint( kwargs["ssh_keys"], kwargs.get("token"), kwargs["names"][0]) if kwargs.get("user_data"): data["user_data"] = kwargs["user_data"] droplets = [] data = api.get_data("droplets/", type=POST, params=data) if data: action_ids = [data["links"]["actions"][0]["id"]] for droplet_json in data["droplets"]: droplet_json["token"] = kwargs["token"] droplet = Droplet(**droplet_json) droplet.action_ids = action_ids droplets.append(droplet) return droplets def __check_actions_in_data(self, data): # reloading actions if actions is provided. if u"actions" in data: self.action_ids = [] for action in data[u'actions']: self.action_ids.append(action[u'id']) def get_data(self, *args, **kwargs): """ Customized version of get_data to perform __check_actions_in_data """ data = super(Droplet, self).get_data(*args, **kwargs) if "type" in kwargs: if kwargs["type"] == POST: self.__check_actions_in_data(data) return data def load(self): """ Fetch data about droplet - use this instead of get_data() """ droplets = self.get_data("droplets/%s" % self.id) droplet = droplets['droplet'] for attr in droplet.keys(): setattr(self, attr, droplet[attr]) for net in self.networks['v4']: if net['type'] == 'private': self.private_ip_address = net['ip_address'] if net['type'] == 'public': self.ip_address = net['ip_address'] if self.networks['v6']: self.ip_v6_address = self.networks['v6'][0]['ip_address'] if "backups" in self.features: self.backups = True else: self.backups = False if "ipv6" in self.features: self.ipv6 = True else: self.ipv6 = False if "private_networking" in self.features: self.private_networking = True else: self.private_networking = False if "tags" in droplets: self.tags = droplets["tags"] return self def _perform_action(self, params, return_dict=True): """ Perform a droplet action. Args: params (dict): parameters of the action Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ action = self.get_data( "droplets/%s/actions/" % self.id, type=POST, params=params ) if return_dict: return action else: action = action[u'action'] return_action = Action(token=self.tokens) # Loading attributes for attr in action.keys(): setattr(return_action, attr, action[attr]) return return_action def power_on(self, return_dict=True): """ Boot up the droplet Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'power_on'}, return_dict) def shutdown(self, return_dict=True): """ shutdown the droplet Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'shutdown'}, return_dict) def reboot(self, return_dict=True): """ restart the droplet Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'reboot'}, return_dict) def power_cycle(self, return_dict=True): """ restart the droplet Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'power_cycle'}, return_dict) def power_off(self, return_dict=True): """ restart the droplet Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'power_off'}, return_dict) def reset_root_password(self, return_dict=True): """ reset the root password Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'password_reset'}, return_dict) def resize(self, new_size_slug, return_dict=True, disk=True): """Resize the droplet to a new size slug. https://developers.digitalocean.com/documentation/v2/#resize-a-droplet Args: new_size_slug (str): name of new size Optional Args: return_dict (bool): Return a dict when True (default), \ otherwise return an Action. disk (bool): If a permanent resize, with disk changes included. Returns dict or Action """ options = {"type": "resize", "size": new_size_slug} if disk: options["disk"] = "true" return self._perform_action(options, return_dict) def take_snapshot(self, snapshot_name, return_dict=True, power_off=False): """Take a snapshot! Args: snapshot_name (str): name of snapshot Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. power_off (bool): Before taking the snapshot the droplet will be turned off with another API call. It will wait until the droplet will be powered off. Returns dict or Action """ if power_off is True and self.status != "off": action = self.power_off(return_dict=False) action.wait() self.load() return self._perform_action( {"type": "snapshot", "name": snapshot_name}, return_dict ) def restore(self, image_id, return_dict=True): """Restore the droplet to an image ( snapshot or backup ) Args: image_id (int): id of image Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action( {"type": "restore", "image": image_id}, return_dict ) def rebuild(self, image_id=None, return_dict=True): """Restore the droplet to an image ( snapshot or backup ) Args: image_id (int): id of image Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ if not image_id: image_id = self.image['id'] return self._perform_action( {"type": "rebuild", "image": image_id}, return_dict ) def enable_backups(self, return_dict=True): """ Enable automatic backups Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'enable_backups'}, return_dict) def disable_backups(self, return_dict=True): """ Disable automatic backups Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'disable_backups'}, return_dict) def destroy(self): """ Destroy the droplet Returns dict """ return self.get_data("droplets/%s" % self.id, type=DELETE) def rename(self, name, return_dict=True): """Rename the droplet Args: name (str): new name Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action( {'type': 'rename', 'name': name}, return_dict ) def enable_private_networking(self, return_dict=True): """ Enable private networking on an existing Droplet where available. Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action( {'type': 'enable_private_networking'}, return_dict ) def enable_ipv6(self, return_dict=True): """ Enable IPv6 on an existing Droplet where available. Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ return self._perform_action({'type': 'enable_ipv6'}, return_dict) def change_kernel(self, kernel, return_dict=True): """Change the kernel to a new one Args: kernel : instance of digitalocean.Kernel.Kernel Optional Args: return_dict (bool): Return a dict when True (default), otherwise return an Action. Returns dict or Action """ if type(kernel) != Kernel: raise BadKernelObject("Use Kernel object") return self._perform_action( {'type': 'change_kernel', 'kernel': kernel.id}, return_dict ) @staticmethod def __get_ssh_keys_id_or_fingerprint(ssh_keys, token, name): """ Check and return a list of SSH key IDs or fingerprints according to DigitalOcean's API. This method is used to check and create a droplet with the correct SSH keys. """ ssh_keys_id = list() for ssh_key in ssh_keys: if type(ssh_key) in [int, type(2 ** 64)]: ssh_keys_id.append(int(ssh_key)) elif type(ssh_key) == SSHKey: ssh_keys_id.append(ssh_key.id) elif type(ssh_key) in [type(u''), type('')]: # ssh_key could either be a fingerprint or a public key # # type(u'') and type('') is the same in python 3 but # different in 2. See: # https://github.com/koalalorenzo/python-digitalocean/issues/80 regexp_of_fingerprint = '([0-9a-fA-F]{2}:){15}[0-9a-fA-F]' match = re.match(regexp_of_fingerprint, ssh_key) if match is not None and match.end() == len(ssh_key) - 1: ssh_keys_id.append(ssh_key) else: key = SSHKey() key.token = token results = key.load_by_pub_key(ssh_key) if results is None: key.public_key = ssh_key key.name = "SSH Key %s" % name key.create() else: key = results ssh_keys_id.append(key.id) else: raise BadSSHKeyFormat( "Droplet.ssh_keys should be a list of IDs, public keys" + " or fingerprints." ) return ssh_keys_id def create(self, *args, **kwargs): """ Create the droplet with object properties. Note: Every argument and parameter given to this method will be assigned to the object. """ for attr in kwargs.keys(): setattr(self, attr, kwargs[attr]) # Provide backwards compatibility if not self.size_slug and self.size: self.size_slug = self.size ssh_keys_id = Droplet.__get_ssh_keys_id_or_fingerprint(self.ssh_keys, self.token, self.name) data = { "name": self.name, "size": self.size_slug, "image": self.image, "region": self.region, "ssh_keys": ssh_keys_id, "backups": bool(self.backups), "ipv6": bool(self.ipv6), "private_networking": bool(self.private_networking), "volumes": self.volumes, "tags": self.tags, "monitoring": bool(self.monitoring), "vpc_uuid": self.vpc_uuid, } if self.user_data: data["user_data"] = self.user_data data = self.get_data("droplets/", type=POST, params=data) if data: self.id = data['droplet']['id'] action_id = data['links']['actions'][0]['id'] self.action_ids = [] self.action_ids.append(action_id) def get_events(self): """ A helper function for backwards compatibility. Calls get_actions() """ return self.get_actions() def get_actions(self): """ Returns a list of Action objects This actions can be used to check the droplet's status """ answer = self.get_data("droplets/%s/actions/" % self.id, type=GET) actions = [] for action_dict in answer['actions']: action = Action(**action_dict) action.token = self.tokens action.droplet_id = self.id action.load() actions.append(action) return actions def get_action(self, action_id): """Returns a specific Action by its ID. Args: action_id (int): id of action """ return Action.get_object( api_token=self.tokens, action_id=action_id ) def get_snapshots(self): """ This method will return the snapshots/images connected to that specific droplet. """ snapshots = list() for id in self.snapshot_ids: snapshot = Image() snapshot.id = id snapshot.token = self.tokens snapshots.append(snapshot) return snapshots def get_kernel_available(self): """ Get a list of kernels available """ kernels = list() data = self.get_data("droplets/%s/kernels/" % self.id) while True: for jsond in data[u'kernels']: kernel = Kernel(**jsond) kernel.token = self.tokens kernels.append(kernel) try: url = data[u'links'][u'pages'].get(u'next') if not url: break data = self.get_data(url) except KeyError: # No links. break return kernels def update_volumes_data(self): """ Trigger volume objects list refresh. When called on a droplet instance, it will take all volumes ids(gathered in initial droplet details collection) and will create list of object of Volume types. Each volume is a separate api call. """ self.volumes = list() for volume_id in self.volume_ids: volume = Volume().get_object(self.tokens, volume_id) self.volumes.append(volume) def __str__(self): return "" % (self.id, self.name)