htsworkflow/util/api.py

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