Source code for libcloud.compute.drivers.vpsnet

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
""" driver
import base64

from libcloud.utils.py3 import b
from libcloud.common.base import JsonResponse, ConnectionUserAndKey
from libcloud.common.types import InvalidCredsError, MalformedResponseError
from libcloud.compute.base import Node, NodeSize, NodeImage, NodeDriver, NodeLocation
from libcloud.compute.types import NodeState
from libcloud.compute.providers import Provider

    import simplejson as json
except ImportError:
    import json

API_VERSION = "api10json"


[docs]class VPSNetResponse(JsonResponse):
[docs] def parse_body(self): try: return super().parse_body() except MalformedResponseError: return self.body
[docs] def success(self): # wrongly uses 406 for invalid auth creds if self.status == 406 or self.status == 403: raise InvalidCredsError() return True
[docs] def parse_error(self): try: errors = super().parse_body()["errors"][0] except MalformedResponseError: return self.body else: return "\n".join(errors)
[docs]class VPSNetConnection(ConnectionUserAndKey): """ Connection class for the driver """ host = API_HOST responseCls = VPSNetResponse allow_insecure = False
[docs] def add_default_headers(self, headers): user_b64 = base64.b64encode(b("{}:{}".format(self.user_id, self.key))) headers["Authorization"] = "Basic %s" % (user_b64.decode("utf-8")) return headers
[docs]class VPSNetNodeDriver(NodeDriver): """ node driver """ type = Provider.VPSNET api_name = "vps_net" name = "" website = "" connectionCls = VPSNetConnection def _to_node(self, vm): if vm["running"]: state = NodeState.RUNNING else: state = NodeState.PENDING n = Node( id=vm["id"], name=vm["label"], state=state, public_ips=[vm.get("primary_ip_address", None)], private_ips=[], extra={"slices_count": vm["slices_count"]}, # Number of nodes consumed by VM driver=self.connection.driver, ) return n def _to_image(self, image, cloud): image = NodeImage( id=image["id"], name="{}: {}".format(cloud, image["label"]), driver=self.connection.driver, ) return image def _to_size(self, num): size = NodeSize( id=num, name="%d Node" % (num,), ram=RAM_PER_NODE * num, disk=DISK_PER_NODE, bandwidth=BANDWIDTH_PER_NODE * num, price=self._get_price_per_node(num) * num, driver=self.connection.driver, ) return size def _get_price_per_node(self, num): single_node_price = self._get_size_price(size_id="1") return num * single_node_price
[docs] def create_node(self, name, image, size, ex_backups_enabled=False, ex_fqdn=None): """Create a new node @inherits: :class:`NodeDriver.create_node` :keyword ex_backups_enabled: Enable automatic backups :type ex_backups_enabled: ``bool`` :keyword ex_fqdn: Fully Qualified domain of the node :type ex_fqdn: ``str`` """ ex_backups_enabled = 1 if ex_backups_enabled else 0 headers = {"Content-Type": "application/json"} request = { "virtual_machine": { "label": name, "fqdn": ex_fqdn or "", "system_template_id":, "backups_enabled": ex_backups_enabled, "slices_required":, } } res = self.connection.request( "/virtual_machines.{}".format(API_VERSION), data=json.dumps(request), headers=headers, method="POST", ) node = self._to_node(res.object["virtual_machine"]) return node
[docs] def reboot_node(self, node): res = self.connection.request( "/virtual_machines/{}/{}.{}".format(, "reboot", API_VERSION), method="POST", ) node = self._to_node(res.object["virtual_machine"]) return True
[docs] def list_sizes(self, location=None): res = self.connection.request("/nodes.{}".format(API_VERSION)) available_nodes = len([size for size in res.object if size["slice"]["virtual_machine_id"]]) sizes = [self._to_size(i) for i in range(1, available_nodes + 1)] return sizes
[docs] def destroy_node(self, node): res = self.connection.request( "/virtual_machines/{}.{}".format(, API_VERSION), method="DELETE" ) return res.status == 200
[docs] def list_nodes(self): res = self.connection.request("/virtual_machines.{}".format(API_VERSION)) return [self._to_node(i["virtual_machine"]) for i in res.object]
[docs] def list_images(self, location=None): res = self.connection.request("/available_clouds.{}".format(API_VERSION)) images = [] for cloud in res.object: label = cloud["cloud"]["label"] templates = cloud["cloud"]["system_templates"] images.extend([self._to_image(image, label) for image in templates]) return images
[docs] def list_locations(self): return [NodeLocation(0, " Western US", "US", self)]