Source code for libcloud.common.gogrid

# 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.

import hashlib
import time

from libcloud.utils.py3 import b

from libcloud.common.types import InvalidCredsError, LibcloudError
from libcloud.common.types import MalformedResponseError
from libcloud.common.base import ConnectionUserAndKey, JsonResponse
from libcloud.compute.base import NodeLocation

HOST = ""
PORTS_BY_SECURITY = {True: 443, False: 80}

__all__ = [

[docs]class GoGridResponse(JsonResponse): def __init__(self, *args, **kwargs): self.driver = BaseGoGridDriver super(GoGridResponse, self).__init__(*args, **kwargs)
[docs] def success(self): if self.status == 403: raise InvalidCredsError("Invalid credentials", self.driver) if self.status == 401: raise InvalidCredsError("API Key has insufficient rights", self.driver) if not self.body: return None try: return self.parse_body()["status"] == "success" except ValueError: raise MalformedResponseError( "Malformed reply", body=self.body, driver=self.driver )
[docs] def parse_error(self): try: return self.parse_body()["list"][0]["message"] except (ValueError, KeyError): return None
[docs]class GoGridConnection(ConnectionUserAndKey): """ Connection class for the GoGrid driver """ host = HOST responseCls = GoGridResponse
[docs] def add_default_params(self, params): params["api_key"] = self.user_id params["v"] = API_VERSION params["format"] = "json" params["sig"] = self.get_signature(self.user_id, self.key) return params
[docs] def get_signature(self, key, secret): """create sig from md5 of key + secret + time""" m = hashlib.md5(b(key + secret + str(int(time.time())))) return m.hexdigest()
[docs] def request( self, action, params=None, data="", headers=None, method="GET", raw=False ): return super(GoGridConnection, self).request( action, params, data, headers, method, raw )
[docs]class GoGridIpAddress(object): """ IP Address """ def __init__(self, id, ip, public, state, subnet): = id self.ip = ip self.public = public self.state = state self.subnet = subnet
[docs]class BaseGoGridDriver(object): """GoGrid has common object model for services they provide, like locations and IP, so keep handling of these things in a single place.""" name = "GoGrid" def _get_ip(self, element): return element.get("ip").get("ip") def _to_ip(self, element): ip = GoGridIpAddress( id=element["id"], ip=element["ip"], public=element["public"], subnet=element["subnet"], state=element["state"]["name"], ) ip.location = self._to_location(element["datacenter"]) return ip def _to_ips(self, object): return [self._to_ip(el) for el in object["list"]] def _to_location(self, element): # pylint: disable=no-member location = NodeLocation( id=element["id"], name=element["name"], country="US", driver=self.connection.driver, ) return location def _to_locations(self, object): return [self._to_location(el) for el in object["list"]]
[docs] def ex_list_ips(self, **kwargs): """Return list of IP addresses assigned to the account. :keyword public: set to True to list only public IPs or False to list only private IPs. Set to None or not specify at all not to filter by type :type public: ``bool`` :keyword assigned: set to True to list only addresses assigned to servers, False to list unassigned addresses and set to None or don't set at all not no filter by state :type assigned: ``bool`` :keyword location: filter IP addresses by location :type location: :class:`NodeLocation` :rtype: ``list`` of :class:`GoGridIpAddress` """ params = {} if "public" in kwargs and kwargs["public"] is not None: params["ip.type"] = {True: "Public", False: "Private"}[kwargs["public"]] if "assigned" in kwargs and kwargs["assigned"] is not None: params["ip.state"] = {True: "Assigned", False: "Unassigned"}[ kwargs["assigned"] ] if "location" in kwargs and kwargs["location"] is not None: params["datacenter"] = kwargs["location"].id # pylint: disable=no-member response = self.connection.request("/api/grid/ip/list", params=params) ips = self._to_ips(response.object) return ips
def _get_first_ip(self, location=None): ips = self.ex_list_ips(public=True, assigned=False, location=location) try: return ips[0].ip except IndexError: # pylint: disable=no-member raise LibcloudError("No public unassigned IPs left", self.driver)