Create batch with premade jobs

Author: hkethi002

api/api.py

@@ -157,12 +157,13 @@ def prefix(path, routes):
 
         # Batch jobs
 
-        route('/batch',                 BatchHandler,   h='get_all',    m=['GET']),
-        route('/batch',                 BatchHandler,                   m=['POST']),
+        route('/batch',                 BatchHandler,   h='get_all',        m=['GET']),
+        route('/batch',                 BatchHandler,                       m=['POST']),
         prefix('/batch', [
-            route('/<:[^/]+>',          BatchHandler,   h='get',        m=['GET']),
-            route('/<:[^/]+>/run',      BatchHandler,   h='run',        m=['POST']),
-            route('/<:[^/]+>/cancel',   BatchHandler,   h='cancel',     m=['POST']),
+            route('/jobs',              BatchHandler,   h='post_with_jobs', m=['POST']),
+            route('/<:[^/]+>',          BatchHandler,   h='get',            m=['GET']),
+            route('/<:[^/]+>/run',      BatchHandler,   h='run',            m=['POST']),
+            route('/<:[^/]+>/cancel',   BatchHandler,   h='cancel',         m=['POST']),
         ]),
 
 

api/jobs/batch.py

@@ -134,84 +134,100 @@ def run(batch_job):
     proposal = batch_job.get('proposal')
     if not proposal:
         raise APIStorageException('The batch job is not formatted correctly.')
-    proposed_inputs = proposal.get('inputs', [])
-    proposed_destinations = proposal.get('destinations', [])
-
-    gear_id = batch_job['gear_id']
-    gear = gears.get_gear(gear_id)
-    gear_name = gear['gear']['name']
-
-    config_ = batch_job.get('config')
-    origin = batch_job.get('origin')
-    tags = proposal.get('tags', [])
-    tags.append('batch')
-
-    if gear.get('category') == 'analysis':
-        analysis_base = proposal.get('analysis', {})
-        if not analysis_base.get('label'):
-            time_now = datetime.datetime.utcnow()
-            analysis_base['label'] = {'label': '{} {}'.format(gear_name, time_now)}
-        an_storage = AnalysisStorage()
-        acq_storage = AcquisitionStorage()
-
-    jobs = []
-    job_ids = []
-
-    job_defaults = {
-        'config':   config_,
-        'gear_id':  gear_id,
-        'tags':     tags,
-        'batch':    str(batch_job.get('_id')),
-        'inputs':   {}
-    }
+    preconstructed_jobs = proposal.get('preconstructed_jobs')
+
+    # If Running a batch from already-constructed jobs
+    if preconstructed_jobs:
+        origin = batch_job.get('origin')
+        jobs = []
+        job_ids = []
+
+        for preconstructed_job in preconstructed_jobs:
+            job = Queue.enqueue_job(preconstructed_job, origin)
+            job_id = job.id_
+            jobs.append(job)
+            job_ids.append(job_id)
 
-    for inputs in proposed_inputs:
+    # Otherwise create jobs from the containers and gear id provided in the proposal
+    else:
+        proposed_inputs = proposal.get('inputs', [])
+        proposed_destinations = proposal.get('destinations', [])
+
+        gear_id = batch_job['gear_id']
+        gear = gears.get_gear(gear_id)
+        gear_name = gear['gear']['name']
 
-        job_map = copy.deepcopy(job_defaults)
-        job_map['inputs'] = inputs
+        config_ = batch_job.get('config')
+        origin = batch_job.get('origin')
+        tags = proposal.get('tags', [])
+        tags.append('batch')
 
         if gear.get('category') == 'analysis':
+            analysis_base = proposal.get('analysis', {})
+            if not analysis_base.get('label'):
+                time_now = datetime.datetime.utcnow()
+                analysis_base['label'] = {'label': '{} {}'.format(gear_name, time_now)}
+            an_storage = AnalysisStorage()
+            acq_storage = AcquisitionStorage()
 
-            analysis = copy.deepcopy(analysis_base)
+        jobs = []
+        job_ids = []
 
-            # Create analysis
-            acquisition_id = inputs.values()[0].get('id')
-            session_id = acq_storage.get_container(acquisition_id, projection={'session':1}).get('session')
-            result = an_storage.create_job_and_analysis('sessions', session_id, analysis, job_map, origin, None)
-            job = result.get('job')
-            job_id = result.get('job_id')
+        job_defaults = {
+            'config':   config_,
+            'gear_id':  gear_id,
+            'tags':     tags,
+            'batch':    str(batch_job.get('_id')),
+            'inputs':   {}
+        }
 
-        else:
+        for inputs in proposed_inputs:
 
-            job = Queue.enqueue_job(job_map, origin)
-            job_id = job.id_
+            job_map = copy.deepcopy(job_defaults)
+            job_map['inputs'] = inputs
 
+            if gear.get('category') == 'analysis':
 
-        jobs.append(job)
-        job_ids.append(job_id)
+                analysis = copy.deepcopy(analysis_base)
 
-    for dest in proposed_destinations:
+                # Create analysis
+                acquisition_id = inputs.values()[0].get('id')
+                session_id = acq_storage.get_container(acquisition_id, projection={'session':1}).get('session')
+                result = an_storage.create_job_and_analysis('sessions', session_id, analysis, job_map, origin, None)
+                job = result.get('job')
+                job_id = result.get('job_id')
 
-        job_map = copy.deepcopy(job_defaults)
-        job_map['destination'] = dest
+            else:
 
-        if gear.get('category') == 'analysis':
+                job = Queue.enqueue_job(job_map, origin)
+                job_id = job.id_
 
-            analysis = copy.deepcopy(analysis_base)
 
-            # Create analysis
-            result = an_storage.create_job_and_analysis('sessions', bson.ObjectId(dest['id']), analysis, job_map, origin, None)
-            job = result.get('job')
-            job_id = result.get('job_id')
+            jobs.append(job)
+            job_ids.append(job_id)
 
-        else:
+        for dest in proposed_destinations:
 
-            job = Queue.enqueue_job(job_map, origin)
-            job_id = job.id_
+            job_map = copy.deepcopy(job_defaults)
+            job_map['destination'] = dest
+
+            if gear.get('category') == 'analysis':
+
+                analysis = copy.deepcopy(analysis_base)
 
+                # Create analysis
+                result = an_storage.create_job_and_analysis('sessions', bson.ObjectId(dest['id']), analysis, job_map, origin, None)
+                job = result.get('job')
+                job_id = result.get('job_id')
 
-        jobs.append(job)
-        job_ids.append(job_id)
+            else:
+
+                job = Queue.enqueue_job(job_map, origin)
+                job_id = job.id_
+
+
+            jobs.append(job)
+            job_ids.append(job_id)
 
     update(batch_job['_id'], {'state': 'running', 'jobs': job_ids})
     return jobs

api/jobs/handlers.py

@@ -699,6 +699,27 @@ def post(self):
 
         return batch_proposal
 
+    @require_login
+    def post_with_jobs(self):
+        """
+        Creates a batch from preconstructed jobs
+        """
+        payload = self.request.json
+        jobs_ = payload.get('jobs', [])
+
+        batch_proposal = {
+            'proposal': {
+                'preconstructed_jobs': jobs_
+            },
+            'origin': self.origin,
+            'state': 'pending',
+            '_id': bson.ObjectId()
+        }
+        batch.insert(batch_proposal)
+        batch_proposal['preconstructed_jobs'] = batch_proposal.pop('proposal')
+
+        return batch_proposal
+
     @require_login
     def run(self, _id):
         """

tests/integration_tests/python/test_batch.py

@@ -84,6 +84,23 @@ def test_batch(data_builder, as_user, as_admin, as_root):
     assert r.ok
     analysis_batch_id = r.json()['_id']
 
+    # create a batch with preconstructed jobs
+    r = as_admin.post('/batch/jobs', json={
+        'jobs': [
+            {
+                'gear_id': gear,
+                'config': { 'two-digit multiple of ten': 20 },
+                'destination': {
+                    'type': 'acquisition',
+                    'id': acquisition
+                },
+                'tags': [ 'test-tag' ]
+            }
+        ]
+    })
+    assert r.ok
+    job_batch_id = r.json()['_id']
+
     # try to get non-existent batch
     r = as_admin.get('/batch/000000000000000000000000')
     assert r.status_code == 404
@@ -97,11 +114,21 @@ def test_batch(data_builder, as_user, as_admin, as_root):
     assert r.ok
     assert r.json()['state'] == 'pending'
 
+    # get batch from jobs
+    r = as_admin.get('/batch/' + job_batch_id)
+    assert r.ok
+    assert r.json()['state'] == 'pending'
+
     # get batch w/ ?jobs=true
     r = as_admin.get('/batch/' + batch_id, params={'jobs': 'true'})
     assert r.ok
     assert 'jobs' in r.json()
 
+    # get job batch w/ ?jobs=true
+    r = as_admin.get('/batch/' + job_batch_id, params={'jobs': 'true'})
+    assert r.ok
+    assert 'jobs' in r.json()
+
     # try to cancel non-running batch
     r = as_admin.post('/batch/' + batch_id + '/cancel')
     assert r.status_code == 400
@@ -134,6 +161,14 @@ def test_batch(data_builder, as_user, as_admin, as_root):
     r = as_admin.get('/batch/' + analysis_batch_id)
     assert r.json()['state'] == 'running'
 
+    # run job batch
+    r = as_admin.post('/batch/' + job_batch_id + '/run')
+    print r.json()
+    assert r.ok
+
+    # test batch.state after calling run
+    r = as_admin.get('/batch/' + job_batch_id)
+    assert r.json()['state'] == 'running'
 
     # Test batch complete
     # create a batch w/ acquisition target and target_context