Source code for libcloud.common.gandi

# 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.
Gandi driver base classes

import time
import hashlib

from libcloud.utils.py3 import b
from libcloud.common.base import ConnectionKey
from libcloud.common.xmlrpc import XMLRPCResponse, XMLRPCConnection

# Global constants

DEFAULT_TIMEOUT = 600  # operation pooling max seconds
DEFAULT_INTERVAL = 20  # seconds between 2

[docs]class GandiException(Exception): """ Exception class for Gandi driver """ def __str__(self): # pylint: disable=unsubscriptable-object return "({}) {}".format(self.args[0], self.args[1]) def __repr__(self): # pylint: disable=unsubscriptable-object return '<GandiException code {} "{}">'.format(self.args[0], self.args[1])
[docs]class GandiResponse(XMLRPCResponse): """ A Base Gandi Response class to derive from. """
[docs]class GandiConnection(XMLRPCConnection, ConnectionKey): """ Connection class for the Gandi driver """ responseCls = GandiResponse host = "" endpoint = "/xmlrpc/" def __init__( self, key, secure=True, timeout=None, retry_delay=None, backoff=None, proxy_url=None, ): # Note: Method resolution order in this case is # XMLRPCConnection -> Connection and Connection doesn't take key as the # first argument so we specify a keyword argument instead. # Previously it was GandiConnection -> ConnectionKey so it worked fine. super().__init__( key=key, secure=secure, timeout=timeout, retry_delay=retry_delay, backoff=backoff, proxy_url=proxy_url, ) self.driver = BaseGandiDriver
[docs] def request(self, method, *args): args = (self.key,) + args return super().request(method, *args)
[docs]class BaseGandiDriver: """ Gandi base driver """ connectionCls = GandiConnection name = "Gandi" # Specific methods for gandi def _wait_operation(self, id, timeout=DEFAULT_TIMEOUT, check_interval=DEFAULT_INTERVAL): """Wait for an operation to succeed""" for i in range(0, timeout, check_interval): try: # pylint: disable=no-member op = self.connection.request("", int(id)).object if op["step"] == "DONE": return True if op["step"] in ["ERROR", "CANCEL"]: return False except (KeyError, IndexError): pass except Exception as e: raise GandiException(1002, e) time.sleep(check_interval) return False
[docs]class BaseObject: """Base class for objects not conventional""" uuid_prefix = "" def __init__(self, id, state, driver): = str(id) if id else None self.state = state self.driver = driver self.uuid = self.get_uuid()
[docs] def get_uuid(self): """Unique hash for this object :return: ``str`` The hash is a function of an SHA1 hash of prefix, the object's ID and its driver which means that it should be unique between all interfaces. TODO : to review >>> from libcloud.compute.drivers.dummy import DummyNodeDriver >>> driver = DummyNodeDriver(0) >>> vif = driver.create_interface() >>> vif.get_uuid() 'd3748461511d8b9b0e0bfa0d4d3383a619a2bb9f' Note, for example, that this example will always produce the same UUID! """ hashstring = "{}:{}:{}".format(self.uuid_prefix,, self.driver.type) return hashlib.sha1(b(hashstring)).hexdigest()
[docs]class IPAddress(BaseObject): """ Provide a common interface for ip addresses """ uuid_prefix = "inet:" def __init__(self, id, state, inet, driver, version=4, extra=None): super().__init__(id, state, driver) self.inet = inet self.version = version self.extra = extra or {} def __repr__(self): return ("<IPAddress: id=%s, address=%s, state=%s, driver=%s ...>") % (, self.inet, self.state,, )
[docs]class NetworkInterface(BaseObject): """ Provide a common interface for network interfaces """ uuid_prefix = "if:" def __init__(self, id, state, mac_address, driver, ips=None, node_id=None, extra=None): super().__init__(id, state, driver) self.mac = mac_address self.ips = ips or {} self.node_id = node_id self.extra = extra or {} def __repr__(self): return ("<Interface: id=%s, mac=%s, state=%s, driver=%s ...>") % (, self.mac, self.state,, )
[docs]class Disk(BaseObject): """ Gandi disk component """ def __init__(self, id, state, name, driver, size, extra=None): super().__init__(id, state, driver) = name self.size = size self.extra = extra or {} def __repr__(self): return ("<Disk: id=%s, name=%s, state=%s, size=%s, driver=%s ...>") % (,, self.state, self.size,, )