WIP: Buffer for qiita updates (#252)

* first pass, part 1

* phase 2

* third pass

* ajax for statuses

* phase four

* sty

* Adds samples with placeholder metadata

* Handler tests

* Unit tests

* conf

* missing section...

* more tweaks

* config...

* sty

* sty

* added a TODO

* Added get_qiita_client test

* A modest mutex for a single threaded process

.travis.yml

@@ -22,7 +22,8 @@ install:
 - travis_retry conda create --yes -n labadmin python=2.7 pip
 - source activate labadmin
 - pip install -U pip
-- pip install -U click natsort coverage coveralls
+- pip install -U click natsort coverage coveralls futures
+- pip install git+https://github.com/qiita-spots/qiita_client
 - travis_retry pip install -U .[test]
 script:
 - git clone https://github.com/biocore/american-gut-web.git ~/build/biocore/american-gut-web
@@ -37,4 +38,3 @@ script:
 - flake8 --ignore E722 knimin setup.py scripts
 after_success:
 - coveralls
-- echo "foo"

knimin/config.txt.example

@@ -34,3 +34,11 @@ PASSKEY =
 USER = test
 PASSWORD = test
 REGISTRATION = test
+
+[qiita]
+QIITA_HOST = test
+QIITA_PORT = 21174
+QIITA_CLIENT_ID = test
+QIITA_CLIENT_SECRET = test
+QIITA_CERT = test
+QIITA_STUDY_ID = 1

knimin/handlers/barcode_util.py

@@ -1,12 +1,42 @@
 #!/usr/bin/env python
 from tornado.web import authenticated
+from tornado.escape import json_encode
+from tornado import gen, concurrent
 from knimin.handlers.base import BaseHandler
 from datetime import datetime
+import requests
+import functools
+from qiita_client import QiitaClient
 
 from knimin import db
 from knimin.lib.constants import survey_type
 from knimin.lib.mail import send_email
 from knimin.handlers.access_decorators import set_access
+from knimin.lib.configuration import config
+
+
+def get_qiita_client():
+    if config.debug:
+        class _mock:
+            def get(self, *args, **kwargs):
+                return {'categories': ['a', 'b', 'c']}
+
+            def http_patch(self, *args, **kwargs):
+                return 'okay'
+
+        qclient = _mock()
+    else:
+        # interface for making HTTP requests against Qiita
+        qclient = QiitaClient(':'.join([config.qiita_host, config.qiita_port]),
+                              config.qiita_client_id,
+                              config.qiita_client_secret,
+                              config.qiita_certificate)
+
+        # we are monkeypatching to use qclient's internal machinery
+        # and to fix the broken HTTP patch
+        qclient.http_patch = functools.partial(qclient._request_retry,
+                                               requests.patch)
+    return qclient
 
 
 class BarcodeUtilHelper(object):
@@ -172,8 +202,66 @@ def _build_email(self, login_user, barcode, email_type,
         return subject, body_message
 
 
+def align_with_qiita_categories(samples, categories):
+    # obviously should instead align to the actual sample metadata...
+    aligned = {}
+    for s in samples:
+        aligned[s] = {c: 'LabControl test' for c in categories}
+    return aligned
+
+
+@set_access(['Scan Barcodes'])
+class PushQiitaHandler(BaseHandler):
+    executor = concurrent.futures.ThreadPoolExecutor(5)
+    study_id = config.qiita_study_id
+    qclient = get_qiita_client()
+
+    @concurrent.run_on_executor
+    def _push_to_qiita(self, study_id, samples):
+        # TODO: add a mutex or block to ensure a single call process at a time
+        cats = self.qclient.get('/api/v1/study/%s/samples/info' % study_id)
+        cats = cats['categories']
+
+        samples = align_with_qiita_categories(samples, cats)
+        data = json_encode(samples)
+
+        return self.qclient.http_patch('/api/v1/study/%s/samples' % study_id,
+                                       data=data)
+
+    @authenticated
+    def get(self):
+        barcodes = db.get_unsent_barcodes_from_qiita_buffer()
+        status = db.get_send_qiita_buffer_status()
+        dat = {'status': status, "barcodes": barcodes}
+        self.write(json_encode(dat))
+        self.finish()
+
+    @authenticated
+    @gen.coroutine
+    def post(self):
+        barcodes = db.get_unsent_barcodes_from_qiita_buffer()
+        if not barcodes:
+            return
+
+        # certainly not a perfect mutex, however tornado is single threaded
+        status = db.get_send_qiita_buffer_status()
+        if status in ['Failed!', 'Pushing...']:
+            return
+
+        db.set_send_qiita_buffer_status("Pushing...")
+
+        try:
+            yield self._push_to_qiita(self.study_id, barcodes)
+        except:  # noqa
+            db.set_send_qiita_buffer_status("Failed!")
+        else:
+            db.mark_barcodes_sent_to_qiita(barcodes)
+            db.set_send_qiita_buffer_status("Idle")
+
+
 @set_access(['Scan Barcodes'])
 class BarcodeUtilHandler(BaseHandler, BarcodeUtilHelper):
+
     @authenticated
     def get(self):
         barcode = self.get_argument('barcode', None)
@@ -293,9 +381,12 @@ def post(self):
 
             new_proj, parent_project = db.getBarcodeProjType(barcode)
         if parent_project == 'American Gut':
+            db.push_barcode_to_qiita_buffer(barcode)
+
             email_msg, ag_update_msg = self.update_ag_barcode(
                 barcode, login_user, login_email, email_type, sent_date,
                 send_mail, sample_date, sample_time, other_text)
+
         self.render("barcode_util.html", div_and_msg=None,
                     barcode_projects=[],
                     parent_project=None,

knimin/lib/configuration.py

@@ -63,6 +63,7 @@ def __init__(self, config_fp=None):
         self._get_tornado(config)
         self._get_email(config)
         self._get_vioscreen(config)
+        self._get_qiita(config)
 
     def _get_main(self, config):
         """Get the configuration of the main section"""
@@ -100,5 +101,13 @@ def _get_vioscreen(self, config):
                                                 config.get('vioscreen',
                                                            'registration'))
 
+    def _get_qiita(self, config):
+        self.qiita_host = config.get('qiita', 'QIITA_HOST')
+        self.qiita_port = config.get('qiita', 'QIITA_PORT')
+        self.qiita_client_id = config.get('qiita', 'QIITA_CLIENT_ID')
+        self.qiita_client_secret = config.get('qiita', 'QIITA_CLIENT_SECRET')
+        self.qiita_certificate = config.get('qiita', 'QIITA_CERT')
+        self.qiita_study_id = config.get('qiita', 'QIITA_STUDY_ID')
+
 
 config = KniminConfig()

knimin/lib/data_access.py

@@ -1402,6 +1402,60 @@ def getAGKitDetails(self, supplied_kit_id):
         else:
             return {}
 
+    def push_barcode_to_qiita_buffer(self, barcode):
+        """Adds barcode to the qiita buffer
+
+        Parameters
+        ----------
+        barcode : str
+            The identifier to stage
+        """
+        sql = """SELECT barcode
+                 FROM project_qiita_buffer"""
+        present = {i[0] for i in self._con.execute_fetchall(sql)}
+
+        if barcode in present:
+            return "Barcode in queue or already sent to Qiita"
+
+        else:
+            sql = """INSERT INTO project_qiita_buffer (barcode)
+                     VALUES (%s)"""
+            self._con.execute(sql, [barcode])
+            return "Barcode inserted"
+
+    def get_send_qiita_buffer_status(self):
+        """Obtain the present status of the Qiita submission buffer"""
+        sql = """SELECT state FROM project_qiita_buffer_status"""
+        return self._con.execute_fetchone(sql)[0]
+
+    def set_send_qiita_buffer_status(self, state):
+        """Obtain the present status of the Qiita submission buffer"""
+        sql = """UPDATE project_qiita_buffer_status
+                 SET state = %s
+                 WHERE id = 0"""
+        self._con.execute(sql, [state])
+
+    def get_unsent_barcodes_from_qiita_buffer(self):
+        """Extract the barcodes that have not been sent to Qiita"""
+        sql = """SELECT barcode
+                 FROM project_qiita_buffer
+                 WHERE pushed_to_qiita='N'"""
+        return [i[0] for i in self._con.execute_fetchall(sql)]
+
+    def mark_barcodes_sent_to_qiita(self, barcodes):
+        """Mark the provided barcodes as sent
+
+        Parameters
+        ----------
+        barcodes : list of str
+            The identifiers to mark a successfully sent to qiita
+        """
+        if barcodes:
+            sql = """UPDATE project_qiita_buffer
+                     SET pushed_to_qiita = 'Y'
+                     WHERE barcode IN %s"""
+            self._con.execute(sql, [tuple(barcodes)])
+
     def add_barcodes_to_kit(self, ag_kit_id, num_barcodes=1):
         """Attaches barcodes to an existing american gut kit
 

knimin/lib/tests/test_configuration.py

@@ -75,6 +75,15 @@ def test_get_tornado(self):
 USER = test
 PASSWORD = test
 REGISTRATION = test
+
+[qiita]
+QIITA_HOST = test
+QIITA_PORT = 21174
+QIITA_CLIENT_ID = test
+QIITA_CLIENT_SECRET = test
+QIITA_CERT = test
+QIITA_STUDY_ID = 1
+
 """
 
 

knimin/lib/tests/test_data_access.py

@@ -24,6 +24,51 @@ def tearDown(self):
         db._clear_table('external_survey_answers', 'ag')
         db._revert_ready(['000023299'])
 
+    def test_push_barcode_to_qiita_buffer(self):
+        db._con.execute('DELETE from barcodes.project_qiita_buffer')
+        db.set_send_qiita_buffer_status('Idle')
+        db.push_barcode_to_qiita_buffer('000004216')
+        db.push_barcode_to_qiita_buffer('000004215')
+        exp = ['000004216', '000004215']
+        obs = db.get_unsent_barcodes_from_qiita_buffer()
+        self.assertEqual(obs, exp)
+
+    def test_get_send_qiita_buffer_status(self):
+        db._con.execute('DELETE from barcodes.project_qiita_buffer')
+        db.set_send_qiita_buffer_status('Idle')
+        exp = 'Idle'
+        obs = db.get_send_qiita_buffer_status()
+        self.assertEqual(obs, exp)
+        db.set_send_qiita_buffer_status('foo')
+        exp = 'foo'
+        obs = db.get_send_qiita_buffer_status()
+        self.assertEqual(obs, exp)
+
+    def test_get_unsent_barcodes_from_qiita_buffer(self):
+        db._con.execute('DELETE from barcodes.project_qiita_buffer')
+        db.set_send_qiita_buffer_status('Idle')
+        db.push_barcode_to_qiita_buffer('000004216')
+        db.push_barcode_to_qiita_buffer('000004215')
+        db._con.execute("""UPDATE barcodes.project_qiita_buffer
+                           SET pushed_to_qiita = 'Y'
+                           WHERE barcode = '000004215'""")
+        obs = db.get_unsent_barcodes_from_qiita_buffer()
+        exp = ['000004216']
+        self.assertEqual(obs, exp)
+
+    def test_set_send_qiita_buffer_status(self):
+        pass  # exercised in test_get_send_qiita_buffer_status
+
+    def test_mark_barcodes_sent_to_qiita(self):
+        db._con.execute('DELETE from barcodes.project_qiita_buffer')
+        db.set_send_qiita_buffer_status('Idle')
+        db.push_barcode_to_qiita_buffer('000004216')
+        db.push_barcode_to_qiita_buffer('000004215')
+        db.mark_barcodes_sent_to_qiita(['000004216'])
+        obs = db.get_unsent_barcodes_from_qiita_buffer()
+        exp = ['000004215']
+        self.assertEqual(obs, exp)
+
     def test_pulldown_third_party(self):
         # Add survey answers
         with open(self.ext_survey_fp, 'rU') as f:

knimin/static/css/qiime.css

@@ -45,6 +45,21 @@ html, body {
 	width: 200px;
 }
 
+#qiita_buffer_window {
+    position: fixed;
+    top: 8px;
+    right: 8px;
+	margin: 8px;
+	background: #fff;
+	opacity: 0.8;
+	padding: 8px;
+	border: 2px black solid;
+	border-radius: 5px;
+	font: 15px normal Arial, Helvetica, sans-serif;
+    word-wrap: break-word;
+	width: 300px;
+}
+
 .login {
 	float:left;
 	margin: 8px;
@@ -294,6 +309,11 @@ h2.verification_text{
 	width: 100%;
 }
 
+h3.qiita_buffer_warning_text{
+    font-style: italic;
+	width: 100%;
+}
+
 #invalid_barcode {
   background-color: #f00;
 }