Skip to content

Instantly share code, notes, and snippets.

@askedrelic

askedrelic/api.py

Created February 17, 2011 00:04
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save askedrelic/830616 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save askedrelic/830616 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. Matt Behrens revised this gist Feb 17, 2011. 2 changed files with 849 additions and 19 deletions.
    835 changes: 835 additions & 0 deletions api.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,835 @@
    #!/usr/bin/python
    # vim:ts=2:sw=2:expandtab
    """
    A Python library to perform low-level Linode API functions.
    Copyright (c) 2010 Timothy J Fontaine <[email protected]>
    Copyright (c) 2010 Josh Wright <[email protected]>
    Copyright (c) 2010 Ryan Tucker <[email protected]>
    Copyright (c) 2008 James C Sinclair <[email protected]>
    Permission is hereby granted, free of charge, to any person
    obtaining a copy of this software and associated documentation
    files (the "Software"), to deal in the Software without
    restriction, including without limitation the rights to use,
    copy, modify, merge, publish, distribute, sublicense, and/or sell
    copies of the Software, and to permit persons to whom the
    Software is furnished to do so, subject to the following
    conditions:
    The above copyright notice and this permission notice shall be
    included in all copies or substantial portions of the Software.
    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
    EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
    OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
    NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
    HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
    WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
    OTHER DEALINGS IN THE SOFTWARE.
    """

    import logging
    import urllib
    import urllib2

    try:
    import json
    except:
    import simplejson as json

    try:
    import VEpycurl
    def vepycurl_request(url, fields, headers):
    return (url, fields, headers)

    def vepycurl_open(request):
    c = VEpycurl.VEpycurl(verifySSL=2)
    url, fields, headers = request
    nh = [ '%s: %s' % (k, v) for k,v in headers.items()]
    c.perform(url, fields, nh)
    return c.results()

    URLOPEN = vepycurl_open
    URLREQUEST = vepycurl_request
    except:
    import warnings
    ssl_message = 'using urllib instead of pycurl, urllib does not verify SSL remote certificates, there is a risk of compromised communication'
    warnings.warn(ssl_message, RuntimeWarning)

    def urllib_request(url, fields, headers):
    fields = urllib.urlencode(fields)
    return urllib2.Request(url, fields, headers)

    URLOPEN = urllib2.urlopen
    URLREQUEST = urllib_request


    class MissingRequiredArgument(Exception):
    """Raised when a required parameter is missing."""

    def __init__(self, value):
    self.value = value
    def __str__(self):
    return repr(self.value)

    class ApiError(Exception):
    """Raised when a Linode API call returns an error.
    Returns:
    [{u'ERRORCODE': Error code number,
    u'ERRORMESSAGE': 'Description of error'}]
    ErrorCodes that can be returned by any method, per Linode API specification:
    0: ok
    1: Bad request
    2: No action was requested
    3: The requested class does not exist
    4: Authentication failed
    5: Object not found
    6: A required property is missing for this action
    7: Property is invalid
    8: A data validation error has occurred
    9: Method Not Implemented
    10: Too many batched requests
    11: RequestArray isn't valid JSON or WDDX
    13: Permission denied
    30: Charging the credit card failed
    31: Credit card is expired
    40: Limit of Linodes added per hour reached
    41: Linode must have no disks before delete
    """

    def __init__(self, value):
    self.value = value
    def __str__(self):
    return repr(self.value)

    class ApiInfo:
    valid_commands = {}
    valid_params = {}

    LINODE_API_URL = 'https://api.linode.com/api/'

    VERSION = '0.0.1'

    class LowerCaseDict(dict):
    def __init__(self, copy=None):
    if copy:
    if isinstance(copy, dict):
    for k,v in copy.items():
    dict.__setitem__(self, k.lower(), v)
    else:
    for k,v in copy:
    dict.__setitem__(self, k.lower(), v)

    def __getitem__(self, key):
    return dict.__getitem__(self, key.lower())

    def __setitem__(self, key, value):
    dict.__setitem__(self, key.lower(), value)

    def __contains__(self, key):
    return dict.__contains__(self, key.lower())

    def get(self, key, def_val=None):
    return dict.get(self, key.lower(), def_val)

    def setdefault(self, key, def_val=None):
    return dict.setdefault(self, key.lower(), def_val)

    def update(self, copy):
    for k,v in copy.items():
    dict.__setitem__(self, k.lower(), v)

    def fromkeys(self, iterable, value=None):
    d = self.__class__()
    for k in iterable:
    dict.__setitem__(d, k.lower(), value)
    return d

    def pop(self, key, def_val=None):
    return dict.pop(self, key.lower(), def_val)

    class Api:
    """Linode API (version 2) client class.
    Instantiate with: Api(), or Api(optional parameters)
    Optional parameters:
    key - Your API key, from "My Profile" in the LPM (default: None)
    batching - Enable batching support (default: False)
    This interfaces with the Linode API (version 2) and receives a response
    via JSON, which is then parsed and returned as a dictionary (or list
    of dictionaries).
    In the event of API problems, raises ApiError:
    api.ApiError: [{u'ERRORCODE': 99,
    u'ERRORMESSAGE': u'Error Message'}]
    If you do not specify a key, the only method you may use is
    user_getapikey(username, password). This will retrieve and store
    the API key for a given user.
    Full documentation on the API can be found from Linode at:
    http://www.linode.com/api/
    """

    def __init__(self, key=None, batching=False):
    self.__key = key
    self.__urlopen = URLOPEN
    self.__request = URLREQUEST
    self.batching = batching
    self.__batch_cache = []

    @staticmethod
    def valid_commands():
    """Returns a list of API commands supported by this class."""
    return list(ApiInfo.valid_commands.keys())

    @staticmethod
    def valid_params():
    """Returns a list of all parameters used by methods of this class."""
    return list(ApiInfo.valid_params.keys())

    def batchFlush(self):
    """Initiates a batch flush. Raises Exception if not in batching mode."""
    if not self.batching:
    raise Exception('Cannot flush requests when not batching')

    s = json.dumps(self.__batch_cache)
    self.__batch_cache = []
    request = { 'api_action' : 'batch', 'api_requestArray' : s }
    return self.__send_request(request)

    def __getattr__(self, name):
    """Return a callable for any undefined attribute and assume it's an API call"""
    def generic_request(*args, **kw):
    request = LowerCaseDict(kw)
    request['api_action'] = name.replace('_', '.')

    if self.batching:
    self.__batch_cache.append(request)
    logging.debug('Batched: %s', json.dumps(request))
    else:
    return self.__send_request(request)
    generic_request.__name__ = name
    return generic_request

    def __send_request(self, request):
    if self.__key:
    request['api_key'] = self.__key
    elif request['api_action'] != 'user.getapikey':
    raise Exception('Must call user_getapikey to fetch key')

    request['api_responseFormat'] = 'json'

    logging.debug('Parmaters '+str(request))
    #request = urllib.urlencode(request)

    headers = {
    'User-Agent': 'LinodePython/'+VERSION,
    }

    req = self.__request(LINODE_API_URL, request, headers)
    response = self.__urlopen(req)
    response = response.read()

    logging.debug('Raw Response: '+response)

    try:
    s = json.loads(response)
    except Exception, ex:
    print(response)
    raise ex

    if isinstance(s, dict):
    s = LowerCaseDict(s)
    if len(s['ERRORARRAY']) > 0:
    if s['ERRORARRAY'][0]['ERRORCODE'] is not 0:
    raise ApiError(s['ERRORARRAY'])
    if s['ACTION'] == 'user.getapikey':
    self.__key = s['DATA']['API_KEY']
    logging.debug('API key is: '+self.__key)
    return s['DATA']
    else:
    return s

    def __api_request(required=[], optional=[], returns=[]):
    """Decorator to define required and optional paramters"""
    for k in required:
    k = k.lower()
    if k not in ApiInfo.valid_params:
    ApiInfo.valid_params[k] = True

    for k in optional:
    k = k.lower()
    if k not in ApiInfo.valid_params:
    ApiInfo.valid_params[k] = True

    def decorator(func):
    if func.__name__ not in ApiInfo.valid_commands:
    ApiInfo.valid_commands[func.__name__] = True

    def wrapper(self, **kw):
    request = LowerCaseDict()
    request['api_action'] = func.__name__.replace('_', '.')

    params = LowerCaseDict(kw)

    for k in required:
    if k not in params:
    raise MissingRequiredArgument(k)

    for k in params:
    request[k] = params[k]

    result = func(self, request)
    if result is not None:
    request = result

    if self.batching:
    self.__batch_cache.append(request)
    logging.debug('Batched: '+ json.dumps(request))
    else:
    return self.__send_request(request)

    wrapper.__name__ = func.__name__
    wrapper.__doc__ = func.__doc__
    wrapper.__dict__.update(func.__dict__)

    if (required or optional) and wrapper.__doc__:
    # Generate parameter documentation in docstring
    if len(wrapper.__doc__.split('\n')) is 1: # one-liners need whitespace
    wrapper.__doc__ += '\n'
    wrapper.__doc__ += '\n Keyword arguments (* = required):\n'
    wrapper.__doc__ += ''.join(['\t *%s\n' % p for p in required])
    wrapper.__doc__ += ''.join(['\t %s\n' % p for p in optional])

    if returns and wrapper.__doc__:
    # we either have a list of dicts or a just plain dict
    if len(wrapper.__doc__.split('\n')) is 1: # one-liners need whitespace
    wrapper.__doc__ += '\n'
    if isinstance(returns, list):
    width = max(len(q) for q in returns[0].keys())
    wrapper.__doc__ += '\n Returns list of dictionaries:\n\t[{\n'
    wrapper.__doc__ += ''.join(['\t %-*s: %s\n'
    % (width, p, returns[0][p]) for p in returns[0].keys()])
    wrapper.__doc__ += '\t }, ...]\n'
    else:
    width = max(len(q) for q in returns.keys())
    wrapper.__doc__ += '\n Returns dictionary:\n\t {\n'
    wrapper.__doc__ += ''.join(['\t %-*s: %s\n'
    % (width, p, returns[p]) for p in returns.keys()])
    wrapper.__doc__ += '\t }\n'

    return wrapper
    return decorator

    @__api_request(optional=['LinodeID'],
    returns=[{u'ALERT_BWIN_ENABLED': '0 or 1',
    u'ALERT_BWIN_THRESHOLD': 'integer (Mb/sec?)',
    u'ALERT_BWOUT_ENABLED': '0 or 1',
    u'ALERT_BWOUT_THRESHOLD': 'integer (Mb/sec?)',
    u'ALERT_BWQUOTA_ENABLED': '0 or 1',
    u'ALERT_BWQUOTA_THRESHOLD': '0..100',
    u'ALERT_CPU_ENABLED': '0 or 1',
    u'ALERT_CPU_THRESHOLD': '0..400 (% CPU)',
    u'ALERT_DISKIO_ENABLED': '0 or 1',
    u'ALERT_DISKIO_THRESHOLD': 'integer (IO ops/sec?)',
    u'BACKUPSENABLED': '0 or 1',
    u'BACKUPWEEKLYDAY': '0..6 (day of week, 0 = Sunday)',
    u'BACKUPWINDOW': 'some integer',
    u'DATACENTERID': 'Datacenter ID',
    u'LABEL': 'linode label',
    u'LINODEID': 'Linode ID',
    u'LPM_DISPLAYGROUP': 'group label',
    u'STATUS': 'Status flag',
    u'TOTALHD': 'available disk (GB)',
    u'TOTALRAM': 'available RAM (MB)',
    u'TOTALXFER': 'available bandwidth (GB/month)',
    u'WATCHDOG': '0 or 1'}])
    def linode_list(self, request):
    """List information about your Linodes.
    Status flag values:
    -2: Boot Failed (not in use)
    -1: Being Created
    0: Brand New
    1: Running
    2: Powered Off
    3: Shutting Down (not in use)
    4: Saved to Disk (not in use)
    """
    pass

    @__api_request(required=['LinodeID'],
    optional=['Label', 'lpm_displayGroup', 'Alert_cpu_enabled',
    'Alert_cpu_threshold', 'Alert_diskio_enabled',
    'Alert_diskio_threshold', 'Alert_bwin_enabled',
    'Alert_bwin_threshold', 'Alert_bwout_enabled',
    'Alert_bwout_threshold', 'Alert_bwquota_enabled',
    'Alert_bwquota_threshold', 'backupWindow',
    'backupWeeklyDay', 'watchdog'],
    returns={u'LinodeID': 'LinodeID'})
    def linode_update(self, request):
    """Update information about, or settings for, a Linode.
    See linode_list.__doc__ for information on parameters.
    """
    pass

    @__api_request(required=['DatacenterID', 'PlanID', 'PaymentTerm'],
    returns={u'LinodeID': 'New Linode ID'})
    def linode_create(self, request):
    """Create a new Linode.
    This will create a billing event.
    """
    pass

    @__api_request(required=['LinodeID'], returns={u'JobID': 'Job ID'})
    def linode_shutdown(self, request):
    """Submit a shutdown job for a Linode.
    On job submission, returns the job ID. Does not wait for job
    completion (see linode_job_list).
    """
    pass

    @__api_request(required=['LinodeID'], optional=['ConfigID'],
    returns={u'JobID': 'Job ID'})
    def linode_boot(self, request):
    """Submit a boot job for a Linode.
    On job submission, returns the job ID. Does not wait for job
    completion (see linode_job_list).
    """
    pass

    @__api_request(required=['LinodeID'], optional=['skipChecks'],
    returns={u'LinodeID': 'Destroyed Linode ID'})
    def linode_delete(self, request):
    """Completely, immediately, and totally deletes a Linode.
    Requires all disk images be deleted first, or that the optional
    skipChecks parameter be set.
    This will create a billing event.
    WARNING: Deleting your last Linode may disable services that
    require a paid account (e.g. DNS hosting).
    """
    pass

    @__api_request(required=['LinodeID'], optional=['ConfigID'],
    returns={u'JobID': 'Job ID'})
    def linode_reboot(self, request):
    """Submit a reboot job for a Linode.
    On job submission, returns the job ID. Does not wait for job
    completion (see linode_job_list).
    """
    pass

    @__api_request(required=['LinodeID', 'PlanID'])
    def linode_resize(self, request):
    """Resize a Linode from one plan to another.
    Immediately shuts the Linode down, charges/credits the account, and
    issues a migration to an appropriate host server.
    """
    pass

    @__api_request(required=['LinodeID'],
    returns=[{u'Comments': 'comments field',
    u'ConfigID': 'Config ID',
    u'DiskList': "',,,,,,,,' disk array",
    u'helper_depmod': '0 or 1',
    u'helper_disableUpdateDB': '0 or 1',
    u'helper_libtls': '0 or 1',
    u'helper_xen': '0 or 1',
    u'KernelID': 'Kernel ID',
    u'Label': 'Profile name',
    u'LinodeID': 'Linode ID',
    u'RAMLimit': 'Max memory (MB), 0 is unlimited',
    u'RootDeviceCustom': '',
    u'RootDeviceNum': 'root partition (1=first, 0=RootDeviceCustom)',
    u'RootDeviceRO': '0 or 1',
    u'RunLevel': "in ['default', 'single', 'binbash'"}])
    def linode_config_list(self, request):
    """Lists all configuration profiles for a given Linode."""
    pass

    @__api_request(required=['LinodeID', 'ConfigID'],
    optional=['KernelID', 'Label', 'Comments', 'RAMLimit',
    'DiskList', 'RunLevel', 'RootDeviceNum',
    'RootDeviceCustom', 'RootDeviceRO',
    'helper_disableUpdateDB', 'helper_xen',
    'helper_depmod'],
    returns={u'ConfigID': 'Config ID'})
    def linode_config_update(self, request):
    """Updates a configuration profile."""
    pass

    @__api_request(required=['LinodeID', 'KernelID', 'Label', 'Disklist'],
    optional=['Comments', 'RAMLimit', 'RunLevel',
    'RootDeviceNum', 'RootDeviceCustom',
    'RootDeviceRO', 'helper_disableUpdateDB',
    'helper_xen', 'helper_depmod'],
    returns={u'ConfigID': 'Config ID'})
    def linode_config_create(self, request):
    """Creates a configuration profile."""
    pass

    @__api_request(required=['LinodeID', 'ConfigID'],
    returns={u'ConfigID': 'Config ID'})
    def linode_config_delete(self, request):
    """Deletes a configuration profile. This does not delete the
    Linode itself, nor its disk images (see linode_disk_delete,
    linode_delete).
    """
    pass

    @__api_request(required=['LinodeID'],
    returns=[{u'CREATE_DT': u'YYYY-MM-DD hh:mm:ss.0',
    u'DISKID': 'Disk ID',
    u'ISREADONLY': '0 or 1',
    u'LABEL': 'Disk label',
    u'LINODEID': 'Linode ID',
    u'SIZE': 'Size of disk (MB)',
    u'STATUS': 'Status flag',
    u'TYPE': "in ['ext3', 'swap', 'raw']",
    u'UPDATE_DT': u'YYYY-MM-DD hh:mm:ss.0'}])
    def linode_disk_list(self, request):
    """Lists all disk images associated with a Linode."""
    pass

    @__api_request(required=['LinodeID', 'DiskID'],
    optional=['Label', 'isReadOnly'],
    returns={u'DiskID': 'Disk ID'})
    def linode_disk_update(self, request):
    """Updates the information about a disk image."""
    pass

    @__api_request(required=['LinodeID', 'Type', 'Size', 'Label'],
    optional=['isReadOnly'],
    returns={u'DiskID': 'Disk ID', u'JobID': 'Job ID'})
    def linode_disk_create(self, request):
    """Submits a job to create a new disk image.
    On job submission, returns the disk ID and job ID. Does not
    wait for job completion (see linode_job_list).
    """
    pass

    @__api_request(required=['LinodeID', 'DiskID'],
    returns={u'DiskID': 'New Disk ID', u'JobID': 'Job ID'})
    def linode_disk_duplicate(self, request):
    """Submits a job to preform a bit-for-bit copy of a disk image.
    On job submission, returns the disk ID and job ID. Does not
    wait for job completion (see linode_job_list).
    """
    pass

    @__api_request(required=['LinodeID', 'DiskID'],
    returns={u'DiskID': 'Deleted Disk ID', u'JobID': 'Job ID'})
    def linode_disk_delete(self, request):
    """Submits a job to delete a disk image.
    WARNING: All data on the disk image will be lost forever.
    On job submission, returns the disk ID and job ID. Does not
    wait for job completion (see linode_job_list).
    """
    pass

    @__api_request(required=['LinodeID', 'DiskID', 'Size'],
    returns={u'DiskID': 'Disk ID', u'JobID': 'Job ID'})
    def linode_disk_resize(self, request):
    """Submits a job to resize a partition.
    On job submission, returns the disk ID and job ID. Does not
    wait for job completion (see linode_job_list).
    """
    pass

    @__api_request(required=['LinodeID', 'DistributionID', 'rootPass', 'Label',
    'Size'],
    optional=['rootSSHKey'],
    returns={u'DiskID': 'New Disk ID', u'JobID': 'Job ID'})
    def linode_disk_createfromdistribution(self, request):
    """Submits a job to create a disk image from a Linode template.
    On job submission, returns the disk ID and job ID. Does not
    wait for job completion (see linode_job_list).
    """
    pass

    @__api_request(required=['LinodeID', 'StackScriptID', 'StackScriptUDFResponses',
    'DistributionID', 'rootPass', 'Label', 'Size'],
    returns={u'DiskID': 'New Disk ID', u'JobID': 'Job ID'})
    def linode_disk_createfromstackscript(self, request):
    """Submits a job to create a disk image from a Linode template.
    On job submission, returns the disk ID and job ID. Does not
    wait for job completion (see linode_job_list). Note: the
    'StackScriptUDFResponses' must be a valid JSON string.
    """
    pass

    @__api_request(required=['LinodeID'],
    returns={u'IPAddressID': 'New IP Address ID'})
    def linode_ip_addprivate(self, request):
    """Assigns a Private IP to a Linode. Returns the IPAddressID
    that was added."""
    pass

    @__api_request(required=['LinodeID'], optional=['IPAddressID'],
    returns=[{u'ISPUBLIC': '0 or 1',
    u'IPADDRESS': '192.168.100.1',
    u'IPADDRESSID': 'IP address ID',
    u'LINODEID': 'Linode ID',
    u'RDNS_NAME': 'reverse.dns.name.here'}])
    def linode_ip_list(self, request):
    """Lists a Linode's IP addresses."""
    pass

    @__api_request(required=['LinodeID'], optional=['pendingOnly', 'JobID'],
    returns=[{u'ACTION': "API action (e.g. u'linode.create')",
    u'DURATION': "Duration spent processing or ''",
    u'ENTERED_DT': 'yyyy-mm-dd hh:mm:ss.0',
    u'HOST_FINISH_DT': "'yyyy-mm-dd hh:mm:ss.0' or ''",
    u'HOST_MESSAGE': 'response from host',
    u'HOST_START_DT': "'yyyy-mm-dd hh:mm:ss.0' or ''",
    u'HOST_SUCCESS': "1 or ''",
    u'JOBID': 'Job ID',
    u'LABEL': 'Description of job',
    u'LINODEID': 'Linode ID'}])
    def linode_job_list(self, request):
    """Returns the contents of the job queue."""
    pass

    @__api_request(optional=['isXen'],
    returns=[{u'ISXEN': '0 or 1',
    u'KERNELID': 'Kernel ID',
    u'LABEL': 'kernel version string'}])
    def avail_kernels(self, request):
    """List available kernels."""
    pass

    @__api_request(returns=[{u'CREATE_DT': 'YYYY-MM-DD hh:mm:ss.0',
    u'DISTRIBUTIONID': 'Distribution ID',
    u'IS64BIT': '0 or 1',
    u'LABEL': 'Description of image',
    u'MINIMAGESIZE': 'MB required to deploy image'}])
    def avail_distributions(self, request):
    """Returns a list of available Linux Distributions."""
    pass

    @__api_request(returns=[{u'DATACENTERID': 'Datacenter ID',
    u'LOCATION': 'City, ST, USA'}])
    def avail_datacenters(self, request):
    """Returns a list of Linode data center facilities."""
    pass

    @__api_request(returns=[{u'DISK': 'Maximum disk allocation (GB)',
    u'LABEL': 'Name of plan', u'PLANID': 'Plan ID',
    u'PRICE': 'Price (US dollars)',
    u'RAM': 'Maximum memory (MB)',
    u'XFER': 'Allowed transfer (GB/mo)',
    u'AVAIL': {u'Datacenter ID': 'Quantity'}}])
    def avail_linodeplans(self, request):
    """Returns a structure of Linode PlanIDs containing PlanIDs, and
    their availability in each datacenter.
    """
    pass

    @__api_request(optional=['StackScriptID', 'DistributionID', 'DistributionVendor',
    'keywords'],
    returns=[{u'CREATE_DT': "'yyyy-mm-dd hh:mm:ss.0'",
    u'DEPLOYMENTSACTIVE': 'The number of Scripts that Depend on this Script',
    u'REV_DT': "'yyyy-mm-dd hh:mm:ss.0'",
    u'DESCRIPTION': 'User defined description of the script',
    u'SCRIPT': 'The actual source of the script',
    u'ISPUBLIC': '0 or 1',
    u'REV_NOTE': 'Comment regarding this revision',
    u'LABEL': 'test',
    u'LATESTREV': 'The number of the latest revision',
    u'DEPLOYMENTSTOTAL': 'Number of times this script has been deployed',
    u'STACKSCRIPTID': 'StackScript ID',
    u'DISTRIBUTIONIDLIST': 'Comma separated list of distributions this script is available'}])
    def avail_stackscripts(self, request):
    """Returns a list of publicly available StackScript.
    """
    pass

    @__api_request(required=['username', 'password'],
    returns={u'API_KEY': 'API key', u'USERNAME': 'Username'})
    def user_getapikey(self, request):
    """Given a username and password, returns the user's API key. The
    key is remembered by this instance for future use.
    Please be advised that this will replace any previous key stored
    by the instance.
    """
    pass

    @__api_request(optional=['DomainID'],
    returns=[{u'STATUS': 'Status flag',
    u'RETRY_SEC': 'SOA Retry field',
    u'DOMAIN': 'Domain name',
    u'DOMAINID': 'Domain ID number',
    u'DESCRIPTION': 'Description',
    u'MASTER_IPS': 'Master nameservers (for slave zones)',
    u'SOA_EMAIL': 'SOA e-mail address (user@domain)',
    u'REFRESH_SEC': 'SOA Refresh field',
    u'TYPE': 'Type of zone (master or slave)',
    u'EXPIRE_SEC': 'SOA Expire field',
    u'TTL_SEC': 'Default TTL'}])
    def domain_list(self, request):
    """Returns a list of domains associated with this account."""
    pass

    @__api_request(required=['DomainID'],
    returns={u'DomainID': 'Domain ID number'})
    def domain_delete(self, request):
    """Deletes a given domain, by domainid."""
    pass

    @__api_request(required=['Domain', 'Type'],
    optional=['SOA_Email', 'Refresh_sec', 'Retry_sec',
    'Expire_sec', 'TTL_sec', 'status', 'master_ips'],
    returns={u'DomainID': 'Domain ID number'})
    def domain_create(self, request):
    """Create a new domain.
    For type='master', SOA_Email is required.
    For type='slave', Master_IPs is required.
    Master_IPs is a comma or semicolon-delimited list of master IPs.
    Status is 1 (Active), 2 (EditMode), or 3 (Off).
    TTL values are rounded up to the nearest valid value:
    300, 3600, 7200, 14400, 28800, 57600, 86400, 172800,
    345600, 604800, 1209600, or 2419200 seconds.
    """
    pass

    @__api_request(required=['DomainID'],
    optional=['Domain', 'Type', 'SOA_Email', 'Refresh_sec',
    'Retry_sec', 'Expire_sec', 'TTL_sec', 'status',
    'master_ips'],
    returns={u'DomainID': 'Domain ID number'})
    def domain_update(self, request):
    """Updates the parameters of a given domain.
    TTL values are rounded up to the nearest valid value:
    300, 3600, 7200, 14400, 28800, 57600, 86400, 172800,
    345600, 604800, 1209600, or 2419200 seconds.
    """
    pass

    @__api_request(required=['DomainID'], optional=['ResourceID'],
    returns=[{u'DOMAINID': 'Domain ID number',
    u'PROTOCOL': 'Protocol (for SRV)',
    u'TTL_SEC': 'TTL for record (0=default)',
    u'WEIGHT': 'Weight (for SRV)',
    u'NAME': 'The hostname or FQDN',
    u'RESOURCEID': 'Resource ID number',
    u'PRIORITY': 'Priority (for MX, SRV)',
    u'TYPE': 'Resource Type (A, MX, etc)',
    u'PORT': 'Port (for SRV)',
    u'TARGET': 'The "right hand side" of the record'}])
    def domain_resource_list(self, request):
    """List the resources associated with a given DomainID."""
    pass

    @__api_request(required=['DomainID', 'Type'],
    optional=['Name', 'Target', 'Priority', 'Weight',
    'Port', 'Protocol', 'TTL_Sec'],
    returns={u'ResourceID': 'Resource ID number'})
    def domain_resource_create(self, request):
    """Creates a resource within a given DomainID.
    TTL values are rounded up to the nearest valid value:
    300, 3600, 7200, 14400, 28800, 57600, 86400, 172800,
    345600, 604800, 1209600, or 2419200 seconds.
    For A and AAAA records, specify Target as "[remote_addr]" to
    use the source IP address of the request as the target, e.g.
    for updating pointers to dynamic IP addresses.
    """
    pass

    @__api_request(required=['DomainID', 'ResourceID'],
    returns={u'ResourceID': 'Resource ID number'})
    def domain_resource_delete(self, request):
    """Deletes a Resource from a Domain."""
    pass

    @__api_request(required=['DomainID', 'ResourceID'],
    optional=['Name', 'Target', 'Priority', 'Weight', 'Port',
    'Protocol', 'TTL_Sec'],
    returns={u'ResourceID': 'Resource ID number'})
    def domain_resource_update(self, request):
    """Updates a domain resource.
    TTL values are rounded up to the nearest valid value:
    300, 3600, 7200, 14400, 28800, 57600, 86400, 172800,
    345600, 604800, 1209600, or 2419200 seconds.
    For A and AAAA records, specify Target as "[remote_addr]" to
    use the source IP address of the request as the target, e.g.
    for updating pointers to dynamic IP addresses.
    """
    pass

    @__api_request(optional=['StackScriptID'],
    returns=[{u'CREATE_DT': "'yyyy-mm-dd hh:mm:ss.0'",
    u'DEPLOYMENTSACTIVE': 'The number of Scripts that Depend on this Script',
    u'REV_DT': "'yyyy-mm-dd hh:mm:ss.0'",
    u'DESCRIPTION': 'User defined description of the script',
    u'SCRIPT': 'The actual source of the script',
    u'ISPUBLIC': '0 or 1',
    u'REV_NOTE': 'Comment regarding this revision',
    u'LABEL': 'test',
    u'LATESTREV': 'The number of the latest revision',
    u'DEPLOYMENTSTOTAL': 'Number of times this script has been deployed',
    u'STACKSCRIPTID': 'StackScript ID',
    u'DISTRIBUTIONIDLIST': 'Comma separated list of distributions this script is available'}])
    def stackscript_list(self, request):
    """List StackScripts you have created.
    """
    pass

    @__api_request(required=['Label', 'DistributionIDList', 'script'],
    optional=['Description', 'isPublic', 'rev_note'],
    returns={'STACKSCRIPTID' : 'ID of the created StackScript'})
    def stackscript_create(self, request):
    """Create a StackScript
    """
    pass

    @__api_request(required=['StackScriptID'],
    optional=['Label', 'Description', 'DistributionIDList',
    'isPublic', 'rev_note', 'script'])
    def stackscript_update(self, request):
    """Update an existing StackScript
    """
    pass

    @__api_request(required=['StackScriptID'])
    def stackscript_delete(self, request):
    """Delete an existing StackScript
    """
    pass

    @__api_request(returns=[{'Parameter' : 'Value'}])
    def test_echo(self, request):
    """Echo back any parameters
    """
    pass

    33 changes: 14 additions & 19 deletions fabfile.py
    View file
    Original file line number Diff line number Diff line change
    @@ -2,31 +2,26 @@

    from fabric.api import *

    ### Before
    def prod():
    """HOST: prod"""
    env.hosts = ['[email protected]']

    def staging():
    """HOST: staging"""
    env.hosts = ['[email protected]']

    ### After
    hosts = {
    'prod' : ['[email protected]'],
    'staging' : ['[email protected]'],
    }

    def _createHost(hostName, hostList):
    def _createHost(hostName, hostSize, hostList):
    moduleObj = sys.modules[__name__]
    def setHost():
    env.hosts = hostList
    #Doc string is readonly during creation?
    setHost.__doc__ = 'HOST: %s ' % hostName
    setHost.__doc__ = 'HOST: %sMB' % hostSize
    setattr(moduleObj, hostName, setHost)

    for host, hostList in hosts.iteritems():
    _createHost(host, hostList)
    def _generateLinodeHosts():
    sys.path.append('.')
    # Import api locally to hide from Fabric
    import api as linodeApi
    LINODE_KEY = "LINODE_KEY_GOES_HERE"
    linode = linodeApi.Api(LINODE_KEY)
    linodeList = [(x['LINODEID'],x['LABEL'], x['TOTALRAM']) for x in linode.linode_list()]
    for server in linodeList:
    ipList = linode.linode_ip_list(LinodeID=int(server[0]))
    publicIps = [x['IPADDRESS'] for x in ipList if x['ISPUBLIC']]
    _createHost(server[1], server[2], [publicIps[0]])
    _generateLinodeHosts()

    def test():
    run('ls')
  2. askedrelic created this gist Feb 17, 2011.
    32 changes: 32 additions & 0 deletions fabfile.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,32 @@
    import sys

    from fabric.api import *

    ### Before
    def prod():
    """HOST: prod"""
    env.hosts = ['[email protected]']

    def staging():
    """HOST: staging"""
    env.hosts = ['[email protected]']

    ### After
    hosts = {
    'prod' : ['[email protected]'],
    'staging' : ['[email protected]'],
    }

    def _createHost(hostName, hostList):
    moduleObj = sys.modules[__name__]
    def setHost():
    env.hosts = hostList
    #Doc string is readonly during creation?
    setHost.__doc__ = 'HOST: %s ' % hostName
    setattr(moduleObj, hostName, setHost)

    for host, hostList in hosts.iteritems():
    _createHost(host, hostList)

    def test():
    run('ls')