Support parameters in workflows (#3333)

* adding support of sample/prep parameters in workflows

* workflows.html changes

* better text in 00~workflows.html01~

* fix qiita_pet test

* change db files

* some cleanup + sample_info PrepTemplateDataHandler

Author: antgonza

qiita_db/handlers/prep_template.py

@@ -108,9 +108,16 @@ def get(self, prep_id):
         dict
             The contents of the prep information keyed by sample id
         """
+        sample_info = self.get_argument('sample_information', False)
+
         with qdb.sql_connection.TRN:
             pt = _get_prep_template(prep_id)
-            response = {'data': pt.to_dataframe().to_dict(orient='index')}
+            if not sample_info:
+                response = {'data': pt.to_dataframe().to_dict(orient='index')}
+            else:
+                ST = qdb.metadata_template.sample_template.SampleTemplate
+                response = {'data': ST(pt.study_id).to_dataframe(
+                    samples=list(pt)).to_dict(orient='index')}
 
         self.write(response)
 

qiita_db/handlers/tests/test_prep_template.py

@@ -130,6 +130,41 @@ def test_get(self):
             'qiita_prep_id': '1'}
         self.assertEqual(obs, exp)
 
+    def test_get_sample_information(self):
+        obs = self.get(
+            '/qiita_db/prep_template/1/data/?sample_information=true',
+            headers=self.header)
+        self.assertEqual(obs.code, 200)
+
+        obs = loads(obs.body)
+        self.assertCountEqual(obs.keys(), ['data'])
+
+        # let's just check that the samples and the keys from the first element
+        # match - this assures us that is the sample_info, the rest is
+        # basically the same as the regular prep_info
+        obs = obs['data']
+        exp = ['1.SKB2.640194', '1.SKM4.640180', '1.SKB3.640195',
+               '1.SKB6.640176', '1.SKD6.640190', '1.SKM6.640187',
+               '1.SKD9.640182', '1.SKM8.640201', '1.SKM2.640199',
+               '1.SKD2.640178', '1.SKB7.640196', '1.SKD4.640185',
+               '1.SKB8.640193', '1.SKM3.640197', '1.SKD5.640186',
+               '1.SKB1.640202', '1.SKM1.640183', '1.SKD1.640179',
+               '1.SKD3.640198', '1.SKB5.640181', '1.SKB4.640189',
+               '1.SKB9.640200', '1.SKM9.640192', '1.SKD8.640184',
+               '1.SKM5.640177', '1.SKM7.640188', '1.SKD7.640191']
+        self.assertCountEqual(list(obs.keys()), exp)
+        exp = ['ph', 'temp', 'depth', 'country', 'texture', 'altitude',
+               'latitude', 'taxon_id', 'elevation', 'env_biome', 'longitude',
+               'tot_nitro', 'host_taxid', 'common_name', 'description',
+               'env_feature', 'env_package', 'sample_type', 'tot_org_carb',
+               'dna_extracted', 'samp_salinity', 'anonymized_name',
+               'host_subject_id', 'scientific_name', 'assigned_from_geo',
+               'season_environment', 'water_content_soil',
+               'collection_timestamp', 'description_duplicate',
+               'physical_specimen_location', 'physical_specimen_remaining',
+               'qiita_study_id']
+        self.assertCountEqual(obs['1.SKB2.640194'].keys(), exp)
+
 
 class PrepTemplateAPItestHandlerTests(OauthTestingBase):
     def test_post(self):

qiita_db/metadata_template/prep_template.py

@@ -328,7 +328,7 @@ def columns_restrictions(self):
         Returns
         -------
         dict
-            The dict of restictions based on the data_type
+            The dict of restrictions based on the data_type
         """
         pt_cols = deepcopy(PREP_TEMPLATE_COLUMNS)
         if self.data_type() in TARGET_GENE_DATA_TYPES:
@@ -850,13 +850,35 @@ def _get_predecessors(workflow, node):
         else:
             starting_job = None
             pt_artifact = self.artifact.artifact_type
-        workflows = [wk for wk in qdb.software.DefaultWorkflow.iter()
-                     if wk.artifact_type == pt_artifact and
-                     pt_dt in wk.data_type]
+
+        workflows = []
+        ST = qdb.metadata_template.sample_template.SampleTemplate
+        for wk in qdb.software.DefaultWorkflow.iter():
+            if wk.artifact_type == pt_artifact and pt_dt in wk.data_type:
+                wk_params = wk.parameters
+                reqs_satisfied = True
+
+                if wk_params['sample']:
+                    df = ST(self.study_id).to_dataframe(samples=list(self))
+                    for k, v in wk_params['sample'].items():
+                        if k not in df.columns or v not in df[k].unique():
+                            reqs_satisfied = False
+
+                if wk_params['prep']:
+                    df = self.to_dataframe()
+                    for k, v in wk_params['prep'].items():
+                        if k not in df.columns or v not in df[k].unique():
+                            reqs_satisfied = False
+
+                if reqs_satisfied:
+                    workflows.append(wk)
+
         if not workflows:
             # raises option a.
             msg = (f'This preparation data type: "{pt_dt}" and/or artifact '
-                   f'type "{pt_artifact}" does not have valid workflows')
+                   f'type "{pt_artifact}" does not have valid workflows; this '
+                   'could be due to required parameters, please check the '
+                   'available workflows.')
             raise ValueError(msg)
         missing_artifacts = dict()
         for wk in workflows:

qiita_db/metadata_template/sample_template.py

@@ -168,7 +168,7 @@ def columns_restrictions(self):
         Returns
         -------
         dict
-            The dict of restictions
+            The dict of restrictions
         """
         return qdb.metadata_template.constants.SAMPLE_TEMPLATE_COLUMNS
 

qiita_db/metadata_template/test/test_prep_template.py

@@ -1369,8 +1369,8 @@ def test_artifact_setter(self):
 
         # here we can test that we can properly create a workflow but we are
         # going to add lot more steps to make it more complex by adding a
-        # couple of new scenarios
-        # 1/2. adds a new path that should be kept separate all the way; this
+        # few more scenarios
+        # 1/4. adds a new path that should be kept separate all the way; this
         #      is to emulate what happens with different trimming (different
         #      default parameter) and deblur (same for each of the previous
         #      steps)
@@ -1385,7 +1385,7 @@ def test_artifact_setter(self):
                 default_workflow_edge_id, parent_output_id, child_input_id)
             VALUES (4, 1, 3)"""
         qdb.sql_connection.perform_as_transaction(sql)
-        # 2/2. adds a new path that should be kept together and then separate;
+        # 2/4. adds a new path that should be kept together and then separate;
         #      this is to simulate what happens with MTX/WGS processing, one
         #      single QC step (together) and 2 separete profilers
         sql = """
@@ -1407,15 +1407,36 @@ def test_artifact_setter(self):
             VALUES (5, 1, 3)
             """
         qdb.sql_connection.perform_as_transaction(sql)
-        # Finally, we need to "activate" the merging scheme values of the
+        # 3/4. we need to "activate" the merging scheme values of the
         # commands so they are actually different:
         # 31->'Pick closed-reference OTUs', 6->'Split libraries FASTQ'
         sql = """
             UPDATE qiita.command_parameter
             SET check_biom_merge = true
             WHERE command_parameter_id IN (31, 6)"""
         qdb.sql_connection.perform_as_transaction(sql)
+        # 4/4. update so we check the prep info file for a valid value
+        sql = """
+            UPDATE qiita.default_workflow
+            SET parameters = '
+                {"sample": {"scientific_name": "1118232"},
+                 "prep": {"center_name": "ANL - 1"}}'::JSONB
+            WHERE default_workflow_id = 1"""
+        qdb.sql_connection.perform_as_transaction(sql)
+
+        # let's check that nothing is found due to the parameters, in specific
+        # "prep": {"center_name": "ANL - 1"}
+        with self.assertRaisesRegex(ValueError, 'This preparation data type: '
+                                    '"16S" and/or artifact type "FASTQ" does '
+                                    'not have valid workflows; this could be '
+                                    'due to required parameters, please check '
+                                    'the available workflows.'):
+            pt.add_default_workflow(qdb.user.User('[email protected]'))
 
+        # now, let's replace the parameters for something fine
+        qdb.software.DefaultWorkflow(1).parameters = {
+            "sample": {"scientific_name": "1118232"},
+            "prep": {"center_name": "ANL"}}
         wk = pt.add_default_workflow(qdb.user.User('[email protected]'))
         self.assertEqual(len(wk.graph.nodes), 5)
         self.assertEqual(len(wk.graph.edges), 3)
@@ -1427,7 +1448,7 @@ def test_artifact_setter(self):
              'Pick closed-reference OTUs'])
 
         # now let's try to generate again and it should fail cause the jobs
-        # are alrady created
+        # are already created
         with self.assertRaisesRegex(ValueError, "Cannot create job because "
                                     "the parameters are the same as jobs"):
             pt.add_default_workflow(qdb.user.User('[email protected]'))

qiita_db/software.py

@@ -2000,3 +2000,46 @@ def graph(self):
                 e = DefaultWorkflowEdge(edge_id)
                 g.add_edge(nodes[p_id], nodes[c_id], connections=e)
         return g
+
+    @property
+    def parameters(self):
+        """Retrieves the parameters that the workflow can be applied to
+
+        Returns
+        ----------
+        dict, dict
+            The dictionary of valid key: value pairs given by the sample or
+            the preparation info file
+        """
+        with qdb.sql_connection.TRN:
+            sql = """SELECT parameters
+                     FROM qiita.default_workflow
+                     WHERE default_workflow_id = %s"""
+            qdb.sql_connection.TRN.add(sql, [self.id])
+            return qdb.sql_connection.TRN.execute_fetchflatten()[0]
+
+    @parameters.setter
+    def parameters(self, values):
+        """Sets the parameters that the workflow can be applied to
+
+        Parameters
+        ----------
+        dict : {'sample': dict, 'prep': dict}
+            dict of dict with the key: value pairs for the 'sample' and 'prep'
+            info files
+
+        Raises
+        ------
+        ValueError
+            if the passed parameter is not a properly formated dict
+        """
+        if not isinstance(values, dict) or \
+                set(values.keys()) != set(['prep', 'sample']):
+            raise ValueError("Improper format for values, should be "
+                             "{'sample': dict, 'prep': dict} ")
+        with qdb.sql_connection.TRN:
+            sql = """UPDATE qiita.default_workflow
+                     SET parameters = %s
+                     WHERE default_workflow_id = %s"""
+            qdb.sql_connection.perform_as_transaction(
+                sql, [dumps(values), self._id])

qiita_db/support_files/patches/89.sql

@@ -1,8 +1,12 @@
 -- Nov 1, 2023
--- add creation_job_id to qiita.prep_template
+-- add slurm/queues changes to support per user_level and analysis parameters
 ALTER TABLE qiita.analysis ADD slurm_reservation VARCHAR DEFAULT '' NOT NULL;
 ALTER TABLE qiita.user_level ADD slurm_parameters VARCHAR DEFAULT '--nice=10000' NOT NULL;
 
 UPDATE qiita.user_level SET slurm_parameters = '--nice=5000' WHERE name = 'admin';
 
 UPDATE qiita.user_level SET slurm_parameters = '' WHERE name = 'wet-lab admin';
+
+-- Nov 22, 2023
+-- add changes to support workflow per sample/prep info specific parameters values
+ALTER TABLE qiita.default_workflow ADD parameters JSONB DEFAULT '{"sample": {}, "prep": {}}'::JSONB NOT NULL;

qiita_db/support_files/qiita-db.dbs

@@ -555,6 +555,9 @@
 			<column name="artifact_type_id" type="bigint" length="19" decimal="0" jt="-5" mandatory="y" >
 				<defo><![CDATA[3]]></defo>
 			</column>
+			<column name="parameters" type="jsonb" jt="2000" mandatory="y" >
+				<defo><![CDATA[{"sample": {}, "prep": {}}]]></defo>
+			</column>
 			<index name="pk_default_workflow" unique="PRIMARY_KEY" >
 				<column name="default_workflow_id" />
 			</index>