htsworkflow/util/api.py

   1 """Common functions for accessing the HTS Workflow REST API
   2 """
   3 import base64
   4 from six.moves import configparser
   5 import random
   6 import logging
   7
   8 # try to deal with python <2.6
   9 try:
  10   import json
  11 except ImportError:
  12   import simplejson as json
  13
  14 import os
  15 from optparse import OptionGroup
  16 from six.moves import urllib
  17
  18 LOGGER = logging.getLogger(__name__)
  19
  20 def add_auth_options(parser):
  21     """Add options OptParser configure authentication options
  22     """
  23     # Load defaults from the config files
  24     config = configparser.SafeConfigParser()
  25     config.read([os.path.expanduser('~/.htsworkflow.ini'),
  26                  '/etc/htsworkflow.ini'
  27                  ])
  28
  29     sequence_archive = None
  30     apiid = None
  31     apikey = None
  32     apihost = None
  33     SECTION = 'sequence_archive'
  34     if config.has_section(SECTION):
  35         sequence_archive = config.get(SECTION, 'sequence_archive',sequence_archive)
  36         sequence_archive = os.path.expanduser(sequence_archive)
  37         apiid = config.get(SECTION, 'apiid', apiid)
  38         apikey = config.get(SECTION, 'apikey', apikey)
  39         apihost = config.get(SECTION, 'host', apihost)
  40
  41     # configuration options
  42     group = OptionGroup(parser, "htsw api authentication")
  43     group.add_option('--apiid', default=apiid, help="Specify API ID")
  44     group.add_option('--apikey', default=apikey, help="Specify API KEY")
  45     group.add_option('--host',  default=apihost,
  46                      help="specify HTSWorkflow host",)
  47     group.add_option('--sequence', default=sequence_archive,
  48                      help="sequence repository")
  49     parser.add_option_group(group)
  50     return parser
  51
  52 def make_auth_from_opts(opts, parser=None):
  53     """Create htsw auth info dictionary from optparse info
  54     """
  55     if opts.host is None or opts.apiid is None or opts.apikey is None:
  56         if parser is not None:
  57             parser.error("Please specify host url, apiid, apikey")
  58         else:
  59             raise RuntimeError("Need host, api id api key")
  60
  61     return {'apiid': opts.apiid, 'apikey': opts.apikey }
  62
  63
  64 def library_url(root_url, library_id):
  65     """
  66     Return the url for retrieving information about a specific library.
  67
  68     Args:
  69       library_id (str): the library id of interest
  70       root_url (str): the root portion of the url, e.g. http://localhost
  71
  72     Returns:
  73       str. The url to use for this REST api.
  74
  75     >>> print library_url('http://localhost', '12345')
  76     http://localhost/samples/library/12345/json
  77
  78     """
  79     url_fragment = '/samples/library/%s/json' % (library_id,)
  80     url = urlparse.urljoin(root_url, url_fragment)
  81
  82     return url
  83
  84
  85 def flowcell_url(root_url, flowcell_id):
  86     """
  87     Return the url for retrieving information about a specific flowcell.
  88
  89     Args:
  90       root_url (str): the root portion of the url, e.g. http://localhost
  91       flowcell_id (str): the flowcell id of interest
  92
  93     Returns:
  94       str. The url to use for this REST api.
  95
  96     >>> print flowcell_url('http://localhost', '1234AAXX')
  97     http://localhost/experiments/config/1234AAXX/json
  98     """
  99     url_fragment = '/experiments/config/%s/json' % (flowcell_id,)
 100     url = urlparse.urljoin(root_url, url_fragment)
 101
 102     return url
 103
 104
 105 def lanes_for_user_url(root_url, username):
 106     """
 107     Return the url for returning all the lanes associated with a username
 108
 109     Args:
 110       username (str): a username in your target filesystem
 111       root_url (str): the root portion of the url, e.g. http://localhost
 112
 113     Returns:
 114       str. The url to use for this REST api.
 115
 116     >>> print lanes_for_user_url('http://localhost', 'diane')
 117     http://localhost/lanes_for/diane/json
 118
 119     """
 120     url_fragment = '/lanes_for/%s/json' % (username,)
 121     url = urlparse.urljoin(root_url, url_fragment)
 122
 123     return url
 124
 125 def retrieve_info(url, apidata):
 126     """
 127     Return a dictionary from the HTSworkflow API
 128     """
 129     try:
 130         apipayload = urllib.parse.urlencode(apidata)
 131         web = urllib.request.urlopen(url, apipayload)
 132     except urllib.request.URLError as e:
 133         if hasattr(e, 'code') and e.code == 404:
 134             LOGGER.info("%s was not found" % (url,))
 135             return None
 136         else:
 137             errmsg = 'URLError: %s' % (str(e))
 138             raise IOError(errmsg)
 139
 140     contents = web.read()
 141     headers = web.info()
 142
 143     return json.loads(contents)
 144
 145 class HtswApi(object):
 146     def __init__(self, root_url, authdata):
 147         self.root_url = root_url
 148         self.authdata = authdata
 149
 150     def get_flowcell(self, flowcellId):
 151         url = flowcell_url(self.root_url, flowcellId)
 152         return retrieve_info(url, self.authdata)
 153
 154     def get_library(self, libraryId):
 155         url = library_url(self.root_url, libraryId)
 156         return retrieve_info(url, self.authdata)
 157
 158     def get_lanes_for_user(self, user):
 159         url = lanes_for_user(self.root_url, user)
 160         return retrieve_info(url, self.authdata)
 161
 162     def get_url(self, url):
 163         return retrieve_info(url, self.authdata)
 164
 165 def make_django_secret_key(size=216):
 166     """return key suitable for use as secret key"""
 167     try:
 168         source = random.SystemRandom()
 169     except AttributeError as e:
 170         source = random.random()
 171     bits = source.getrandbits(size)
 172     chars = []
 173     while bits > 0:
 174         byte = bits & 0xff
 175         chars.append(chr(byte))
 176         bits >>= 8
 177     return base64.encodestring("".join(chars)).strip()
 178
 179 if __name__ == "__main__":
 180     from optparse import OptionParser
 181     from pprint import pprint
 182     parser = OptionParser()
 183     parser = add_auth_options(parser)
 184     parser.add_option('--flowcell', default=None)
 185     parser.add_option('--library', default=None)
 186
 187     opts, args = parser.parse_args()
 188     apidata =  make_auth_from_opts(opts)
 189
 190     api = HtswApi(opts.host, apidata)
 191
 192     if opts.flowcell is not None:
 193         pprint(api.get_flowcell(opts.flowcell))
 194     if opts.library is not None:
 195         pprint(api.get_library(opts.library))
 196