@@ -1311,8 +1311,12 @@ def _complete_artifact_definition(self, artifact_data):
1311
1311
data_type = data_type , name = job_params ['name' ])
1312
1312
self ._set_status ('success' )
1313
1313
1314
- if list (self .children ):
1315
- self ._update_and_launch_children ({atype : artifact .id })
1314
+ # we need to update the children jobs to replace the input
1315
+ # for the newly created artifact via the validator
1316
+ for c in self .children :
1317
+ self ._helper_update_children ({atype : artifact .id })
1318
+
1319
+ self ._update_and_launch_children ()
1316
1320
1317
1321
def _complete_artifact_transformation (self , artifacts_data ):
1318
1322
"""Performs the needed steps to complete an artifact transformation job
@@ -1682,6 +1686,42 @@ def validator_jobs(self):
1682
1686
for jid in qdb .sql_connection .TRN .execute_fetchflatten ():
1683
1687
yield ProcessingJob (jid )
1684
1688
1689
+ def _helper_update_children (self , new_map ):
1690
+ ready = []
1691
+ sql = """SELECT command_parameters, pending
1692
+ FROM qiita.processing_job
1693
+ WHERE processing_job_id = %s"""
1694
+ sql_update = """UPDATE qiita.processing_job
1695
+ SET command_parameters = %s,
1696
+ pending = %s
1697
+ WHERE processing_job_id = %s"""
1698
+ sql_link = """INSERT INTO qiita.artifact_processing_job
1699
+ (artifact_id, processing_job_id)
1700
+ VALUES (%s, %s)"""
1701
+
1702
+ for c in self .children :
1703
+ qdb .sql_connection .TRN .add (sql , [c .id ])
1704
+ params , pending = qdb .sql_connection .TRN .execute_fetchflatten ()
1705
+ for pname , out_name in pending [self .id ].items ():
1706
+ a_id = new_map [out_name ]
1707
+ params [pname ] = str (a_id )
1708
+ del pending [self .id ]
1709
+ # Link the input artifact with the child job
1710
+ qdb .sql_connection .TRN .add (sql_link , [a_id , c .id ])
1711
+
1712
+ # Force to insert a NULL in the DB if pending is empty
1713
+ pending = pending if pending else None
1714
+ qdb .sql_connection .TRN .add (sql_update ,
1715
+ [dumps (params ), pending , c .id ])
1716
+ qdb .sql_connection .TRN .execute ()
1717
+
1718
+ if pending is None :
1719
+ # The child already has all the parameters
1720
+ # Add it to the ready list
1721
+ ready .append (c )
1722
+
1723
+ return ready
1724
+
1685
1725
def _update_children (self , mapping ):
1686
1726
"""Updates the children of the current job to populate the input params
1687
1727
@@ -1695,7 +1735,6 @@ def _update_children(self, mapping):
1695
1735
list of qiita_db.processing_job.ProcessingJob
1696
1736
The list of childrens that are ready to be submitted
1697
1737
"""
1698
- ready = []
1699
1738
with qdb .sql_connection .TRN :
1700
1739
sql = """SELECT command_output_id, name
1701
1740
FROM qiita.command_output
@@ -1705,37 +1744,7 @@ def _update_children(self, mapping):
1705
1744
res = qdb .sql_connection .TRN .execute_fetchindex ()
1706
1745
new_map = {name : mapping [oid ] for oid , name in res }
1707
1746
1708
- sql = """SELECT command_parameters, pending
1709
- FROM qiita.processing_job
1710
- WHERE processing_job_id = %s"""
1711
- sql_update = """UPDATE qiita.processing_job
1712
- SET command_parameters = %s,
1713
- pending = %s
1714
- WHERE processing_job_id = %s"""
1715
- sql_link = """INSERT INTO qiita.artifact_processing_job
1716
- (artifact_id, processing_job_id)
1717
- VALUES (%s, %s)"""
1718
- for c in self .children :
1719
- qdb .sql_connection .TRN .add (sql , [c .id ])
1720
- params , pending = qdb .sql_connection .TRN .execute_fetchflatten ()
1721
- for pname , out_name in pending [self .id ].items ():
1722
- a_id = new_map [out_name ]
1723
- params [pname ] = str (a_id )
1724
- del pending [self .id ]
1725
- # Link the input artifact with the child job
1726
- qdb .sql_connection .TRN .add (sql_link , [a_id , c .id ])
1727
-
1728
- # Force to insert a NULL in the DB if pending is empty
1729
- pending = pending if pending else None
1730
- qdb .sql_connection .TRN .add (sql_update ,
1731
- [dumps (params ), pending , c .id ])
1732
- qdb .sql_connection .TRN .execute ()
1733
-
1734
- if pending is None :
1735
- # The child already has all the parameters
1736
- # Add it to the ready list
1737
- ready .append (c )
1738
- return ready
1747
+ return self ._helper_update_children (new_map )
1739
1748
1740
1749
def _update_and_launch_children (self , mapping ):
1741
1750
"""Updates the children of the current job to populate the input params
0 commit comments