htsworkflow/util/api.py

   1 """
   2 Common functions for accessing the HTS Workflow REST API
   3
   4 """
   5 # try to deal with python <2.6
   6 try:
   7   import json
   8 except ImportError:
   9   import simplejson as json
  10
  11 import urllib
  12 import urllib2
  13 import urlparse
  14
  15 def library_url(root_url, library_id):
  16     """
  17     Return the url for retrieving information about a specific library.
  18
  19     Args:
  20       library_id (str): the library id of interest
  21       root_url (str): the root portion of the url, e.g. http://localhost
  22
  23     Returns:
  24       str. The url to use for this REST api.
  25
  26     >>> print library_url('http://localhost', '12345')
  27     http://localhost/samples/library/12345/json
  28
  29     """
  30     url_fragment = '/samples/library/%s/json' % (library_id,)
  31     url = urlparse.urljoin(root_url, url_fragment)
  32
  33     return url
  34
  35
  36 def flowcell_url(root_url, flowcell_id):
  37     """
  38     Return the url for retrieving information about a specific flowcell.
  39
  40     Args:
  41       root_url (str): the root portion of the url, e.g. http://localhost
  42       flowcell_id (str): the flowcell id of interest
  43
  44     Returns:
  45       str. The url to use for this REST api.
  46
  47     >>> print flowcell_url('http://localhost', '1234AAXX')
  48     http://localhost/experiments/config/1234AAXX/json
  49     """
  50     url_fragment = '/experiments/config/%s/json' % (flowcell_id,)
  51     url = urlparse.urljoin(root_url, url_fragment)
  52
  53     return url
  54
  55
  56 def lanes_for_user_url(root_url, username):
  57     """
  58     Return the url for returning all the lanes associated with a username
  59
  60     Args:
  61       username (str): a username in your target filesystem
  62       root_url (str): the root portion of the url, e.g. http://localhost
  63
  64     Returns:
  65       str. The url to use for this REST api.
  66
  67     >>> print lanes_for_user_url('http://localhost', 'diane')
  68     http://localhost/lanes_for/diane/json
  69
  70     """
  71     url_fragment = '/lanes_for/%s/json' % (username,)
  72     url = urlparse.urljoin(root_url, url_fragment)
  73
  74     return url
  75
  76 def retrieve_info(url, apidata):
  77     """
  78     Return a dictionary from the HTSworkflow API
  79     """
  80     try:
  81         apipayload = urllib.urlencode(apidata)
  82         web = urllib2.urlopen(url, apipayload)
  83     except urllib2.URLError, e:
  84         if e.code == 404:
  85             logging.info("%s was not found" % (url,))
  86             return None
  87         else:
  88             errmsg = 'URLError: %d %s' % (e.code, e.msg)
  89             raise IOError(errmsg)
  90
  91     contents = web.read()
  92     headers = web.info()
  93
  94     return json.loads(contents)