htsworkflow/util/api.py

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