mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Galaxy AMQP Listener, first pass
Added a check for testing completed requests in Requests functional tests
This commit is contained in:
Ramkrishna Chakrabarty
@@ -98,7 +98,6 @@ class Requests( BaseController ):
|
||||
else:
|
||||
self.request_grid.default_filter = dict(state=kwargs['show_filter'], deleted=False)
|
||||
self.request_grid.show_filter = kwargs.get('show_filter', trans.app.model.Request.states.SUBMITTED)
|
||||
self.__update_request_state(trans)
|
||||
# Render the list view
|
||||
return self.request_grid( trans, **kwargs )
|
||||
@web.expose
|
||||
@@ -980,12 +979,6 @@ class Requests( BaseController ):
|
||||
request_id=request.id,
|
||||
msg='Bar codes has been saved for this request',
|
||||
messagetype='done'))
|
||||
|
||||
def __update_request_state(self, trans):
|
||||
requests = trans.app.model.Request.query.filter_by(deleted=False,
|
||||
state=trans.app.model.Request.states.SUBMITTED)
|
||||
for request in requests:
|
||||
self.__set_request_state(request)
|
||||
|
||||
def __set_request_state(self, request):
|
||||
# check if all the samples of the current request are in the final state
|
||||
|
||||
@@ -121,7 +121,7 @@ class Grid( object ):
|
||||
return None
|
||||
def build_initial_query( self, session ):
|
||||
return session.query( self.model_class )
|
||||
def apply_default_filter( self, trans, query ):
|
||||
def apply_default_filter( self, trans, query, **kwargs):
|
||||
return query
|
||||
|
||||
class GridColumn( object ):
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
#!/bin/sh
|
||||
|
||||
python scripts/galaxy_messaging/amqp_consumer.py
|
||||
@@ -0,0 +1,68 @@
|
||||
from amqplib import client_0_8 as amqp
|
||||
import ConfigParser
|
||||
import sys
|
||||
import optparse
|
||||
import xml.dom.minidom
|
||||
from galaxydb_interface import GalaxyDbInterface
|
||||
|
||||
galaxy_config_file = 'universe_wsgi.ini'
|
||||
global dbconnstr
|
||||
|
||||
def get_value(dom, tag_name):
|
||||
'''
|
||||
This method extracts the tag value from the xml message
|
||||
'''
|
||||
nodelist = dom.getElementsByTagName(tag_name)[0].childNodes
|
||||
rc = ""
|
||||
for node in nodelist:
|
||||
if node.nodeType == node.TEXT_NODE:
|
||||
rc = rc + node.data
|
||||
return rc
|
||||
|
||||
def recv_callback(msg):
|
||||
#print 'Received: ' + msg.body + ' from channel #' + str(msg.channel.channel_id)
|
||||
dom = xml.dom.minidom.parseString(msg.body)
|
||||
barcode = get_value(dom, 'barcode')
|
||||
state = get_value(dom, 'state')
|
||||
print barcode, state
|
||||
# update the galaxy db
|
||||
galaxy = GalaxyDbInterface(dbconnstr)
|
||||
sample_id = galaxy.get_sample_id(field_name='bar_code', value=barcode)
|
||||
if sample_id == -1:
|
||||
print 'Invalid barcode.'
|
||||
return
|
||||
galaxy.change_state(sample_id, state)
|
||||
|
||||
def main():
|
||||
config = ConfigParser.ConfigParser()
|
||||
config.read(galaxy_config_file)
|
||||
global dbconnstr
|
||||
dbconnstr = config.get("app:main", "database_connection")
|
||||
amqp_config = {}
|
||||
for option in config.options("galaxy:amqp"):
|
||||
amqp_config[option] = config.get("galaxy:amqp", option)
|
||||
print amqp_config
|
||||
conn = amqp.Connection(host=amqp_config['host']+":"+amqp_config['port'],
|
||||
userid=amqp_config['userid'],
|
||||
password=amqp_config['password'],
|
||||
virtual_host=amqp_config['virtual_host'],
|
||||
insist=False)
|
||||
chan = conn.channel()
|
||||
chan.queue_declare(queue=amqp_config['queue'], durable=True, exclusive=True, auto_delete=False)
|
||||
chan.exchange_declare(exchange=amqp_config['exchange'], type="direct", durable=True, auto_delete=False,)
|
||||
chan.queue_bind(queue=amqp_config['queue'],
|
||||
exchange=amqp_config['exchange'],
|
||||
routing_key=amqp_config['routing_key'])
|
||||
|
||||
chan.basic_consume(queue=amqp_config['queue'],
|
||||
no_ack=True,
|
||||
callback=recv_callback,
|
||||
consumer_tag="testtag")
|
||||
while True:
|
||||
chan.wait()
|
||||
chan.basic_cancel("testtag")
|
||||
chan.close()
|
||||
conn.close()
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,141 @@
|
||||
#/usr/bin/python
|
||||
|
||||
from sqlalchemy import *
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from datetime import datetime, timedelta
|
||||
import sys
|
||||
import optparse
|
||||
import os
|
||||
import time
|
||||
import logging
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
log = logging.getLogger( 'GalaxyDbInterface' )
|
||||
|
||||
class GalaxyDbInterface(object):
|
||||
|
||||
def __init__(self, dbstr):
|
||||
self.dbstr = dbstr
|
||||
self.db_engine = create_engine(self.dbstr)
|
||||
# self.db_engine.echo = True
|
||||
self.metadata = MetaData(self.db_engine)
|
||||
self.session = sessionmaker(bind=self.db_engine)
|
||||
self.event_table = Table('sample_event', self.metadata, autoload=True )
|
||||
self.sample_table = Table('sample', self.metadata, autoload=True )
|
||||
self.request_table = Table('request', self.metadata, autoload=True )
|
||||
self.state_table = Table('sample_state', self.metadata, autoload=True )
|
||||
|
||||
def get_sample_id(self, field_name='bar_code', value=None):
|
||||
if not value:
|
||||
return -1
|
||||
sample_id = -1
|
||||
if field_name =='name':
|
||||
stmt = select(columns=[self.sample_table.c.id],
|
||||
whereclause=self.sample_table.c.name==value)
|
||||
result = stmt.execute()
|
||||
sample_id = result.fetchone()[0]
|
||||
elif field_name == 'bar_code':
|
||||
stmt = select(columns=[self.sample_table.c.id],
|
||||
whereclause=self.sample_table.c.bar_code==value)
|
||||
result = stmt.execute()
|
||||
x = result.fetchone()
|
||||
if x:
|
||||
sample_id = x[0]
|
||||
log.debug('Sample ID: %i' % sample_id)
|
||||
return sample_id
|
||||
log.warning('This sample %s %s does not belong to any sample in the database.' % (field_name, value))
|
||||
return -1
|
||||
|
||||
def current_state(self, sample_id):
|
||||
'''
|
||||
This method returns the current state of the sample for the given sample_id
|
||||
'''
|
||||
stmt = select(columns=[self.event_table.c.sample_state_id],
|
||||
whereclause=self.event_table.c.sample_id==sample_id,
|
||||
order_by=self.event_table.c.update_time.desc())
|
||||
result = stmt.execute()
|
||||
all_states = result.fetchall()
|
||||
current_state_id = all_states[0][0]
|
||||
return current_state_id
|
||||
|
||||
def all_possible_states(self, sample_id):
|
||||
subsubquery = select(columns=[self.sample_table.c.request_id],
|
||||
whereclause=self.sample_table.c.id==sample_id)
|
||||
self.request_id = subsubquery.execute().fetchall()[0][0]
|
||||
log.debug('REQUESTID: %i' % self.request_id)
|
||||
subquery = select(columns=[self.request_table.c.request_type_id],
|
||||
whereclause=self.request_table.c.id==self.request_id)
|
||||
request_type_id = subquery.execute().fetchall()[0][0]
|
||||
log.debug('REQUESTTYPEID: %i' % request_type_id)
|
||||
query = select(columns=[self.state_table.c.id, self.state_table.c.name],
|
||||
whereclause=self.state_table.c.request_type_id==request_type_id,
|
||||
order_by=self.state_table.c.id.asc())
|
||||
states = query.execute().fetchall()
|
||||
log.debug('POSSIBLESTATES: '+ str(states))
|
||||
return states
|
||||
|
||||
def change_state(self, sample_id, new_state=None):
|
||||
'''
|
||||
This method changes the state of the sample to the the 'new_state'
|
||||
'''
|
||||
if not new_state:
|
||||
return
|
||||
new_state_id = -1
|
||||
# find the state_id for this new state in the list of possible states
|
||||
possible_states = self.all_possible_states(sample_id)
|
||||
for state_id, state_name in possible_states:
|
||||
if new_state == state_name:
|
||||
new_state_id = state_id
|
||||
if new_state_id == -1:
|
||||
return
|
||||
log.debug('Updating sample_id %i state to %s' % (sample_id, new_state))
|
||||
d = timedelta(hours=4)
|
||||
i = self.event_table.insert()
|
||||
i.execute(update_time=datetime.now()+d,
|
||||
create_time=datetime.now()+d,
|
||||
sample_id=sample_id,
|
||||
sample_state_id=int(new_state_id),
|
||||
comment='bar code scanner')
|
||||
# if all the samples for this request are in the final state
|
||||
# then change the request state to 'Complete'
|
||||
result = select(columns=[self.sample_table.c.id],
|
||||
whereclause=self.sample_table.c.request_id==self.request_id).execute()
|
||||
sample_id_list = result.fetchall()
|
||||
request_complete = True
|
||||
for sid in sample_id_list:
|
||||
current_state_id = self.current_state(sid[0])
|
||||
if current_state_id != possible_states[-1][0]:
|
||||
request_complete = False
|
||||
break
|
||||
if request_complete:
|
||||
request_state = 'Complete'
|
||||
else:
|
||||
request_state = 'Submitted'
|
||||
log.debug('Updating request_id %i state to "%s"' % (self.request_id, request_state))
|
||||
d = timedelta(hours=4)
|
||||
i = self.request_table.update(whereclause=self.request_table.c.id==self.request_id,
|
||||
values={self.request_table.c.state: request_state})
|
||||
i.execute()
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print '''This file should not be run directly. To start the Galaxy AMQP Listener:
|
||||
%sh run_galaxy_listener.sh'''
|
||||
# dbstr = 'postgres://postgres:postgres@localhost/galaxy_ft'
|
||||
#
|
||||
# parser = optparse.OptionParser()
|
||||
# parser.add_option('-n', '--name', help='name of the sample field', dest='name', \
|
||||
# action='store', default='bar_code')
|
||||
# parser.add_option('-v', '--value', help='value of the sample field', dest='value', \
|
||||
# action='store')
|
||||
# parser.add_option('-s', '--state', help='new state of the sample', dest='state', \
|
||||
# action='store')
|
||||
# (opts, args) = parser.parse_args()
|
||||
#
|
||||
# gs = GalaxyDbInterface(dbstr)
|
||||
# sample_id = gs.get_sample_id(field_name=opts.name, value=opts.value)
|
||||
# gs.change_state(sample_id, opts.state)
|
||||
|
||||
|
||||
|
||||
@@ -247,6 +247,8 @@ class TestFormsAndRequests( TwillTestCase ):
|
||||
self.home()
|
||||
request_one.refresh()
|
||||
# check if the request's state is now set to 'complete'
|
||||
self.visit_url('%s/requests_admin/list?show_filter=Complete' % self.url)
|
||||
self.check_page_for_string( request_one.name )
|
||||
assert request_one.state is not request_one.states.COMPLETE, "The state of the request '%s' should be set to '%s'" % ( request_one.name, request_one.states.COMPLETE )
|
||||
# def test_40_admin_create_request_on_behalf_of_regular_user( self ):
|
||||
# """Testing creating and submitting a request as an admin on behalf of a regular user"""
|
||||
|
||||
@@ -240,3 +240,22 @@ ucsc_table_direct1 = local:///
|
||||
ucsc_table_direct_archaea1 = local:///
|
||||
ucsc_table_direct_test1 = local:///
|
||||
upload1 = local:///
|
||||
|
||||
|
||||
# Galaxy Message Queue
|
||||
# Galaxy uses AMQ protocol to receive messages from external sources like
|
||||
# bar code scanners. Galaxy has been tested against RabbitMQ AMQP implementation.
|
||||
# For Galaxy to receive messages from a message queue the RabbitMQ server has
|
||||
# to be set up with a user account and other parameters listed below. The 'host'
|
||||
# and 'port' fields should point to where the RabbitMQ server is running.
|
||||
|
||||
#[galaxy:amqp]
|
||||
#host = 127.0.0.1
|
||||
#port = 5672
|
||||
#userid = galaxy
|
||||
#password = galaxy
|
||||
#virtual_host = galaxy_messaging_engine
|
||||
#queue = galaxy_queue
|
||||
#exchange = galaxy_exchange
|
||||
#routing_key = bar_code_scanner
|
||||
|
||||
|
||||
Reference in New Issue
Block a user