Merge pull request #356 from scitran/session-jobs

Lookup jobs at session level

api/dao/containerutil.py

@@ -86,6 +86,11 @@ class ContainerReference(object):
         if type not in CONT_TYPES:
             raise Exception('Container type must be one of {}'.format(CONT_TYPES))
 
+        if not isinstance(type, basestring):
+           raise Exception('Container type must be of type str')
+        if not isinstance(id, basestring):
+           raise Exception('Container id must be of type str')
+
         self.type = type
         self.id   = id
 

api/dao/hierarchy.py

@@ -58,6 +58,21 @@ def get_container(cont_name, _id):
         '_id': _id,
     })
 
+def get_children(cont_name, _id):
+    """
+    Given a container name and id, return all children of that object in the hierarchy
+    """
+    cid = bson.ObjectId(_id)
+    if cont_name == 'session':
+        return config.db.acquisitions.find({'session': cid})
+    elif cont_name == 'project':
+        return config.db.sessions.find({'project': cid})
+    elif cont_name == 'group':
+        # groups do not use ObjectIds
+        return config.db.projects.find({'group':_id})
+    else:
+        raise ValueError('Children can only be listed from group, project or session level')
+
 def propagate_changes(cont_name, _id, query, update):
     """
     Propagates changes down the heirarchy tree.

api/handlers/containerhandler.py

@@ -9,7 +9,7 @@ from .. import config
 from .. import debuginfo
 from .. import validators
 from ..auth import containerauth, always_ok
-from ..dao import APIStorageException, containerstorage, containerutil, noop
+from ..dao import APIStorageException, containerstorage, containerutil, noop, hierarchy
 from ..types import Origin
 from ..jobs.queue import Queue
 
@@ -189,7 +189,7 @@ class ContainerHandler(base.RequestHandler):
 
             .. sourcecode:: http
 
-                GET /api/acquisitions/3/jobs?states=pending&states=failed&tags=a&tags=b HTTP/1.1
+                GET /api/sessions/3/jobs?states=pending&states=failed&tags=a&tags=b HTTP/1.1
                 Host: demo.flywheel.io
                 Accept: */*
 
@@ -225,14 +225,32 @@ class ContainerHandler(base.RequestHandler):
 
         self.config = self.container_handler_configurations[cont_name]
         self.storage = self.config['storage']
+        if cont_name != 'sessions':
+            self.abort(400, 'Can only request jobs at the session level.')
+
         c = self._get_container(cid)
         permchecker = self._get_permchecker(c)
         result = permchecker(noop)('GET', cid)
 
-        cr	 = containerutil.ContainerReference(cont_name[:-1], cid)
+        children = hierarchy.get_children('session', cid)
+        id_array = [str(c['_id']) for c in children]
+        cont_array = [containerutil.ContainerReference('acquisition', cid) for cid in id_array]
+
         states = self.request.GET.getall('states')
         tags   = self.request.GET.getall('tags')
-        return Queue.search(cr, states=states, tags=tags)
+        jobs = Queue.search(cont_array, states=states, tags=tags)
+
+        response = {}
+        for j in jobs:
+            log.debug(j)
+            inputs = j.get('inputs', {})
+            for k,v in inputs.items():
+                if v['type'] == 'acquisition' and v['id'] in id_array:
+                    if response.get(v['id']) is not None:
+                        response[v['id']].append(j)
+                    else:
+                        response[v['id']] = [j]
+        return response
 
     def get_all(self, cont_name, par_cont_name=None, par_id=None):
         self.config = self.container_handler_configurations[cont_name]

api/jobs/queue.py

@@ -167,21 +167,37 @@ class Queue(object):
         return result
 
     @staticmethod
-    def search(container, states=None, tags=None):
+    def search(containers, states=None, tags=None):
         """
-        Search the queue for jobs that mention a specific container and (optionally) match some set of states or tags.
+        Search the queue for jobs that mention at least one of a set of containers and (optionally) match some set of states or tags.
+        Currently, all containers must be of the same type.
+
+        @param containers: an array of ContainerRefs
+        @param states: an array of strings
+        @param tags: an array of strings
         """
 
+        # Limitation: container types must match.
+        type1 = containers[0].type
+        for container in containers:
+            if container.type != type1:
+                raise Exception('All containers passed to Queue.search must be of the same type')
+
+        array_string = str([ x.id for x in containers ])
+
         filter = """
+            var ids = $ids$
             for (var key in this['inputs']) {
                 var ct = this['inputs'][key]['type']
                 var ci = this['inputs'][key]['id']
-                if (ct === '$cT$' && ci == '$cI$') { return true }
+                if (ct === '$cT$' && ids.indexOf(ci) >= 0){ return true }
             }
-        """.replace('$cT$', container.type).replace('$cI$', container.id)
+        """.replace('$cT$', type1).replace('$ids$', array_string)
 
         query = { "$where": filter }
 
+        log.debug(query)
+
         if states is not None and len(states) > 0:
             query['state'] = {"$in": states}