Datasets
Models
Diff of
/qiita_db/artifact.py
[000000]
..
[879b32]
Switch to side-by-side view
--- a +++ b/qiita_db/artifact.py @@ -0,0 +1,1746 @@ +# ----------------------------------------------------------------------------- +# Copyright (c) 2014--, The Qiita Development Team. +# +# Distributed under the terms of the BSD 3-clause License. +# +# The full license is in the file LICENSE, distributed with this software. +# ----------------------------------------------------------------------------- +from itertools import chain +from datetime import datetime +from os import remove +from os.path import isfile, relpath +from shutil import rmtree +from collections import namedtuple +from json import dumps +from qiita_db.util import create_nested_path + +import networkx as nx + +import qiita_db as qdb + +from qiita_core.qiita_settings import qiita_config + + +TypeNode = namedtuple('TypeNode', ['id', 'job_id', 'name', 'type']) + + +class Artifact(qdb.base.QiitaObject): + r"""Any kind of file (or group of files) stored in the system and its + attributes + + Attributes + ---------- + timestamp + processing_parameters + visibility + artifact_type + data_type + can_be_submitted_to_ebi + can_be_submitted_to_vamps + is_submitted_to_vamps + filepaths + parents + prep_template + ebi_run_accession + study + has_human + + Methods + ------- + create + delete + being_deleted_by + archive + + See Also + -------- + qiita_db.QiitaObject + """ + _table = "artifact" + + @classmethod + def iter(cls): + """Iterate over all artifacts in the database + + Returns + ------- + generator + Yields a `Artifact` object for each artifact in the database, + in order of ascending artifact_id + """ + with qdb.sql_connection.TRN: + sql = """SELECT artifact_id FROM qiita.{} + ORDER BY artifact_id""".format(cls._table) + qdb.sql_connection.TRN.add(sql) + + ids = qdb.sql_connection.TRN.execute_fetchflatten() + + for id_ in ids: + yield Artifact(id_) + + @classmethod + def iter_by_visibility(cls, visibility): + r"""Iterator over the artifacts with the given visibility + + Parameters + ---------- + visibility : str + The visibility level + + Returns + ------- + generator of qiita_db.artifact.Artifact + The artifacts available in the system with the given visibility + """ + with qdb.sql_connection.TRN: + sql = """SELECT artifact_id + FROM qiita.artifact + JOIN qiita.visibility USING (visibility_id) + WHERE visibility = %s + ORDER BY artifact_id""" + qdb.sql_connection.TRN.add(sql, [visibility]) + for a_id in qdb.sql_connection.TRN.execute_fetchflatten(): + yield cls(a_id) + + @staticmethod + def types(): + """Returns list of all artifact types available and their descriptions + + Returns + ------- + list of list of str + The artifact type and description of the artifact type, in the form + [[artifact_type, description, can_be_submitted_to_ebi, + can_be_submitted_to_vamps, is_user_uploadable], ...] + """ + with qdb.sql_connection.TRN: + sql = """SELECT artifact_type, description, + can_be_submitted_to_ebi, + can_be_submitted_to_vamps, is_user_uploadable + FROM qiita.artifact_type + ORDER BY artifact_type""" + qdb.sql_connection.TRN.add(sql) + return qdb.sql_connection.TRN.execute_fetchindex() + + @staticmethod + def create_type(name, description, can_be_submitted_to_ebi, + can_be_submitted_to_vamps, is_user_uploadable, + filepath_types): + """Creates a new artifact type in the system + + Parameters + ---------- + name : str + The artifact type name + description : str + The artifact type description + can_be_submitted_to_ebi : bool + Whether the artifact type can be submitted to EBI or not + can_be_submitted_to_vamps : bool + Whether the artifact type can be submitted to VAMPS or not + is_user_uploadable : bool + Whether the artifact type can be raw: upload directly to qiita + filepath_types : list of (str, bool) + The list filepath types that the new artifact type supports, and + if they're required or not in an artifact instance of this type + + Raises + ------ + qiita_db.exceptions.QiitaDBDuplicateError + If an artifact type with the same name already exists + """ + with qdb.sql_connection.TRN: + sql = """SELECT EXISTS( + SELECT * + FROM qiita.artifact_type + WHERE artifact_type=%s)""" + qdb.sql_connection.TRN.add(sql, [name]) + if qdb.sql_connection.TRN.execute_fetchlast(): + raise qdb.exceptions.QiitaDBDuplicateError( + 'artifact type', 'name: %s' % name) + sql = """INSERT INTO qiita.artifact_type + (artifact_type, description, can_be_submitted_to_ebi, + can_be_submitted_to_vamps, is_user_uploadable) + VALUES (%s, %s, %s, %s, %s) + RETURNING artifact_type_id""" + qdb.sql_connection.TRN.add( + sql, [name, description, can_be_submitted_to_ebi, + can_be_submitted_to_vamps, is_user_uploadable]) + at_id = qdb.sql_connection.TRN.execute_fetchlast() + sql = """INSERT INTO qiita.artifact_type_filepath_type + (artifact_type_id, filepath_type_id, required) + VALUES (%s, %s, %s)""" + sql_args = [ + [at_id, qdb.util.convert_to_id(fpt, 'filepath_type'), req] + for fpt, req in filepath_types] + qdb.sql_connection.TRN.add(sql, sql_args, many=True) + + # When creating a type is expected that a new mountpoint is created + # for that type, note that we are going to check if there is an + # extra path for the mountpoint, which is useful for the test + # environment + qc = qiita_config + mp = relpath(qc.working_dir, qc.base_data_dir).replace( + 'working_dir', '') + mp = mp + name if mp != '/' and mp != '' else name + sql = """INSERT INTO qiita.data_directory + (data_type, mountpoint, subdirectory, active) + VALUES (%s, %s, %s, %s)""" + qdb.sql_connection.TRN.add(sql, [name, mp, True, True]) + + # We are intersted in the dirpath + create_nested_path(qdb.util.get_mountpoint(name)[0][1]) + + qdb.sql_connection.TRN.execute() + + @classmethod + def copy(cls, artifact, prep_template): + """Creates a copy of `artifact` and attaches it to `prep_template` + + Parameters + ---------- + artifact : qiita_db.artifact.Artifact + Artifact to copy from + prep_template : qiita_db.metadata_template.prep_template.PrepTemplate + The prep template to attach the new artifact to + + Returns + ------- + qiita_db.artifact.Artifact + A new instance of Artifact + """ + with qdb.sql_connection.TRN: + visibility_id = qdb.util.convert_to_id("sandbox", "visibility") + atype = artifact.artifact_type + atype_id = qdb.util.convert_to_id(atype, "artifact_type") + dtype_id = qdb.util.convert_to_id( + prep_template.data_type(), "data_type") + sql = """INSERT INTO qiita.artifact ( + generated_timestamp, visibility_id, artifact_type_id, + data_type_id, submitted_to_vamps) + VALUES (%s, %s, %s, %s, %s) + RETURNING artifact_id""" + sql_args = [datetime.now(), visibility_id, atype_id, dtype_id, + False] + qdb.sql_connection.TRN.add(sql, sql_args) + a_id = qdb.sql_connection.TRN.execute_fetchlast() + + # Associate the artifact with the prep template + instance = cls(a_id) + prep_template.artifact = instance + + # Associate the artifact with the study + sql = """INSERT INTO qiita.study_artifact (study_id, artifact_id) + VALUES (%s, %s)""" + sql_args = [prep_template.study_id, a_id] + qdb.sql_connection.TRN.add(sql, sql_args) + + # Associate the artifact with the preparation information + sql = """INSERT INTO qiita.preparation_artifact ( + prep_template_id, artifact_id) + VALUES (%s, %s)""" + sql_args = [prep_template.id, a_id] + qdb.sql_connection.TRN.add(sql, sql_args) + + # Associate the artifact with its filepaths + filepaths = [(x['fp'], x['fp_type']) for x in artifact.filepaths] + fp_ids = qdb.util.insert_filepaths( + filepaths, a_id, atype, copy=True) + sql = """INSERT INTO qiita.artifact_filepath + (artifact_id, filepath_id) + VALUES (%s, %s)""" + sql_args = [[a_id, fp_id] for fp_id in fp_ids] + qdb.sql_connection.TRN.add(sql, sql_args, many=True) + qdb.sql_connection.TRN.execute() + + return instance + + @classmethod + def create(cls, filepaths, artifact_type, name=None, prep_template=None, + parents=None, processing_parameters=None, move_files=True, + analysis=None, data_type=None): + r"""Creates a new artifact in the system + + The parameters depend on how the artifact was generated: + - If the artifact was uploaded by the user, the parameter + `prep_template` should be provided and the parameters `parents`, + `processing_parameters` and `analysis` should not be provided. + - If the artifact was generated by processing one or more + artifacts, the parameters `parents` and `processing_parameters` + should be provided and the parameters `prep_template` and + `analysis` should not be provided. + - If the artifact is the initial artifact of the analysis, the + parameters `analysis` and `data_type` should be provided and the + parameters `prep_template`, `parents` and `processing_parameters` + should not be provided. + + Parameters + ---------- + filepaths : iterable of tuples (str, int) + A list of 2-tuples in which the first element is the artifact + file path and the second one is the file path type id + artifact_type : str + The type of the artifact + name : str, optional + The artifact's name + prep_template : qiita_db.metadata_template.PrepTemplate, optional + If the artifact is being uploaded by the user, the prep template + to which the artifact should be linked to. If not provided, + `parents` or `analysis` should be provided. + parents : iterable of qiita_db.artifact.Artifact, optional + The list of artifacts from which the new artifact has been + generated. If not provided, `prep_template` or `analysis` + should be provided. + processing_parameters : qiita_db.software.Parameters, optional + The processing parameters used to generate the new artifact + from `parents`. It is required if `parents` is provided. It should + not be provided if `processing_parameters` is not provided. + move_files : bool, optional + If False the files will not be moved but copied + analysis : qiita_db.analysis.Analysis, optional + If the artifact is the inital artifact of an analysis, the analysis + to which the artifact belongs to. If not provided, `prep_template` + or `parents` should be provided. + data_type : str + The data_type of the artifact in the `analysis`. It is required if + `analysis` is provided. It should not be provided if `analysis` is + not provided. + + Returns + ------- + qiita_db.artifact.Artifact + A new instance of Artifact + + Raises + ------ + QiitaDBArtifactCreationError + If `filepaths` is not provided + If both `parents` and `prep_template` are provided + If none of `parents` and `prep_template` are provided + If `parents` is provided but `processing_parameters` is not + If both `prep_template` and `processing_parameters` is provided + If not all the artifacts in `parents` belong to the same study + + Notes + ----- + The visibility of the artifact is set by default to `sandbox` if + prep_template is passed but if parents is passed we will inherit the + most closed visibility. + The timestamp of the artifact is set by default to `datetime.now()`. + The value of `submitted_to_vamps` is set by default to `False`. + """ + # We need at least one file + if not filepaths: + raise qdb.exceptions.QiitaDBArtifactCreationError( + "at least one filepath is required.") + + # Check that the combination of parameters is correct + counts = (int(bool(parents or processing_parameters)) + + int(prep_template is not None) + + int(bool(analysis or data_type))) + if counts != 1: + # More than one parameter has been provided + raise qdb.exceptions.QiitaDBArtifactCreationError( + "One and only one of parents, prep template or analysis must " + "be provided") + elif bool(parents) != bool(processing_parameters): + # When provided, parents and processing parameters both should be + # provided (this is effectively doing an XOR) + raise qdb.exceptions.QiitaDBArtifactCreationError( + "When provided, both parents and processing parameters should " + "be provided") + elif bool(analysis) and not bool(data_type): + # When provided, analysis and data_type both should be + # provided (this is effectively doing an XOR) + raise qdb.exceptions.QiitaDBArtifactCreationError( + "When provided, both analysis and data_type should " + "be provided") + + # There are three different ways of creating an Artifact, but all of + # them execute a set of common operations. Declare functions to avoid + # code duplication. These functions should not be used outside of the + # CREATE OR REPLACE FUNCTION, hence declaring them here + def _common_creation_steps(atype, cmd_id, data_type, cmd_parameters): + gen_timestamp = datetime.now() + visibility_id = qdb.util.convert_to_id("sandbox", "visibility") + atype_id = qdb.util.convert_to_id(atype, "artifact_type") + dtype_id = qdb.util.convert_to_id(data_type, "data_type") + # Create the artifact row in the artifact table + sql = """INSERT INTO qiita.artifact + (generated_timestamp, command_id, data_type_id, + command_parameters, visibility_id, + artifact_type_id, submitted_to_vamps) + VALUES (%s, %s, %s, %s, %s, %s, %s) + RETURNING artifact_id""" + sql_args = [gen_timestamp, cmd_id, dtype_id, + cmd_parameters, visibility_id, atype_id, False] + qdb.sql_connection.TRN.add(sql, sql_args) + a_id = qdb.sql_connection.TRN.execute_fetchlast() + qdb.sql_connection.TRN.execute() + + return cls(a_id) + + def _associate_with_study(instance, study_id, prep_template_id): + # Associate the artifact with the study + sql = """INSERT INTO qiita.study_artifact + (study_id, artifact_id) + VALUES (%s, %s)""" + sql_args = [study_id, instance.id] + qdb.sql_connection.TRN.add(sql, sql_args) + sql = """INSERT INTO qiita.preparation_artifact + (prep_template_id, artifact_id) + VALUES (%s, %s)""" + sql_args = [prep_template_id, instance.id] + qdb.sql_connection.TRN.add(sql, sql_args) + qdb.sql_connection.TRN.execute() + + def _associate_with_analysis(instance, analysis_id): + # Associate the artifact with the analysis + sql = """INSERT INTO qiita.analysis_artifact + (analysis_id, artifact_id) + VALUES (%s, %s)""" + sql_args = [analysis_id, instance.id] + qdb.sql_connection.perform_as_transaction(sql, sql_args) + + with qdb.sql_connection.TRN: + if parents: + dtypes = {p.data_type for p in parents} + # If an artifact has parents, it can be either from the + # processing pipeline or the analysis pipeline. Decide which + # one here + studies = {p.study for p in parents} + analyses = {p.analysis for p in parents} + studies.discard(None) + analyses.discard(None) + studies = {s.id for s in studies} + analyses = {a.id for a in analyses} + + # The first 2 cases should never happen, but it doesn't hurt + # to check them + len_studies = len(studies) + len_analyses = len(analyses) + if len_studies > 0 and len_analyses > 0: + raise qdb.exceptions.QiitaDBArtifactCreationError( + "All the parents from an artifact should be either " + "from the analysis pipeline or all from the processing" + " pipeline") + elif len_studies > 1 or len_studies > 1: + raise qdb.exceptions.QiitaDBArtifactCreationError( + "Parents from multiple studies/analyses provided. " + "Analyses: %s. Studies: %s." + % (', '.join(analyses), ', '.join(studies))) + elif len_studies == 1: + # This artifact is part of the processing pipeline + study_id = studies.pop() + # In the processing pipeline, artifacts can have only + # one dtype + if len(dtypes) > 1: + raise qdb.exceptions.QiitaDBArtifactCreationError( + "parents have multiple data types: %s" + % ", ".join(dtypes)) + + instance = _common_creation_steps( + artifact_type, processing_parameters.command.id, + dtypes.pop(), processing_parameters.dump()) + _associate_with_study( + instance, study_id, parents[0].prep_templates[0].id) + else: + # This artifact is part of the analysis pipeline + analysis_id = analyses.pop() + # In the processing pipeline, artifact parents can have + # more than one data type + data_type = ("Multiomic" + if len(dtypes) > 1 else dtypes.pop()) + instance = _common_creation_steps( + artifact_type, processing_parameters.command.id, + data_type, processing_parameters.dump()) + _associate_with_analysis(instance, analysis_id) + + # Associate the artifact with its parents + sql = """INSERT INTO qiita.parent_artifact + (artifact_id, parent_id) + VALUES (%s, %s)""" + sql_args = [(instance.id, p.id) for p in parents] + qdb.sql_connection.TRN.add(sql, sql_args, many=True) + + # inheriting visibility + visibilities = {a.visibility for a in instance.parents} + # set based on the "lowest" visibility + if 'sandbox' in visibilities: + instance.visibility = 'sandbox' + elif 'private' in visibilities: + instance.visibility = 'private' + else: + instance._set_visibility('public') + + elif prep_template: + # This artifact is uploaded by the user in the + # processing pipeline + instance = _common_creation_steps( + artifact_type, None, prep_template.data_type(), None) + # Associate the artifact with the prep template + prep_template.artifact = instance + # Associate the artifact with the study + _associate_with_study( + instance, prep_template.study_id, prep_template.id) + else: + # This artifact is an initial artifact of an analysis + instance = _common_creation_steps( + artifact_type, None, data_type, None) + # Associate the artifact with the analysis + if bool(analysis): + analysis.add_artifact(instance) + + # Associate the artifact with its filepaths + fp_ids = qdb.util.insert_filepaths( + filepaths, instance.id, artifact_type, + move_files=move_files, copy=(not move_files)) + sql = """INSERT INTO qiita.artifact_filepath + (artifact_id, filepath_id) + VALUES (%s, %s)""" + sql_args = [[instance.id, fp_id] for fp_id in fp_ids] + qdb.sql_connection.TRN.add(sql, sql_args, many=True) + + if name: + instance.name = name + + return instance + + @classmethod + def delete(cls, artifact_id): + r"""Deletes an artifact from the system with its children + + Parameters + ---------- + artifact_id : int + The parent artifact to be removed + + Raises + ------ + QiitaDBArtifactDeletionError + If the artifacts are public + If the artifacts have been analyzed + If the artifacts have been submitted to EBI + If the artifacts have been submitted to VAMPS + """ + with qdb.sql_connection.TRN: + # This will fail if the artifact with id=artifact_id doesn't exist + instance = cls(artifact_id) + + # Check if the artifact is public + if instance.visibility == 'public': + raise qdb.exceptions.QiitaDBArtifactDeletionError( + artifact_id, "it is public") + + all_artifacts = list(set(instance.descendants.nodes())) + all_artifacts.reverse() + all_ids = tuple([a.id for a in all_artifacts]) + + # Check if this or any of the children have been analyzed + sql = """SELECT email, analysis_id + FROM qiita.analysis + WHERE analysis_id IN ( + SELECT DISTINCT analysis_id + FROM qiita.analysis_sample + WHERE artifact_id IN %s)""" + qdb.sql_connection.TRN.add(sql, [all_ids]) + analyses = qdb.sql_connection.TRN.execute_fetchindex() + if analyses: + analyses = '\n'.join( + ['Analysis id: %s, Owner: %s' % (aid, email) + for email, aid in analyses]) + raise qdb.exceptions.QiitaDBArtifactDeletionError( + artifact_id, 'it or one of its children has been ' + 'analyzed by: \n %s' % analyses) + + # Check if the artifacts have been submitted to EBI + for a in all_artifacts: + if a.can_be_submitted_to_ebi and a.ebi_run_accessions: + raise qdb.exceptions.QiitaDBArtifactDeletionError( + artifact_id, "Artifact %d has been submitted to " + "EBI" % a.id) + + # Check if the artifacts have been submitted to VAMPS + for a in all_artifacts: + if a.can_be_submitted_to_vamps and a.is_submitted_to_vamps: + raise qdb.exceptions.QiitaDBArtifactDeletionError( + artifact_id, "Artifact %d has been submitted to " + "VAMPS" % a.id) + + # Check if there is a job queued, running, waiting or + # in_construction that will use/is using the artifact + sql = """SELECT processing_job_id + FROM qiita.artifact_processing_job + JOIN qiita.processing_job USING (processing_job_id) + JOIN qiita.processing_job_status + USING (processing_job_status_id) + WHERE artifact_id IN %s + AND processing_job_status IN ( + 'queued', 'running', 'waiting')""" + qdb.sql_connection.TRN.add(sql, [all_ids]) + jobs = qdb.sql_connection.TRN.execute_fetchflatten() + if jobs: + # if the artifact has active jobs we need to raise an error + # but we also need to check that if it's only 1 job, that the + # job is not the delete_artifact actual job + raise_error = True + job_name = qdb.processing_job.ProcessingJob( + jobs[0]).command.name + if len(jobs) == 1 and job_name == 'delete_artifact': + raise_error = False + if raise_error: + raise qdb.exceptions.QiitaDBArtifactDeletionError( + artifact_id, "there is a queued/running job that " + "uses this artifact or one of it's children") + + # We can now remove the artifacts + filepaths = [f for a in all_artifacts for f in a.filepaths] + study = instance.study + + # Delete any failed/successful job that had the artifact as input + sql = """SELECT processing_job_id + FROM qiita.artifact_processing_job + WHERE artifact_id IN %s""" + qdb.sql_connection.TRN.add(sql, [all_ids]) + job_ids = tuple(qdb.sql_connection.TRN.execute_fetchflatten()) + + if job_ids: + sql = """DELETE FROM qiita.artifact_processing_job + WHERE artifact_id IN %s""" + qdb.sql_connection.TRN.add(sql, [all_ids]) + + # Delete the entry from the artifact_output_processing_job table + sql = """DELETE FROM qiita.artifact_output_processing_job + WHERE artifact_id IN %s""" + qdb.sql_connection.TRN.add(sql, [all_ids]) + + # Detach the artifact from its filepaths + sql = """DELETE FROM qiita.artifact_filepath + WHERE artifact_id IN %s""" + qdb.sql_connection.TRN.add(sql, [all_ids]) + + # If the first artifact to be deleted, instance, doesn't have + # parents and study is not None (None means is an analysis), we + # move the files to the uploads folder. We also need + # to nullify the column in the prep template table + if not instance.parents and study is not None: + qdb.util.move_filepaths_to_upload_folder(study.id, + filepaths) + # there are cases that an artifact would not be linked to a + # study + pt_ids = [tuple([pt.id]) for a in all_artifacts + for pt in a.prep_templates] + if pt_ids: + sql = """UPDATE qiita.prep_template + SET artifact_id = NULL + WHERE prep_template_id IN %s""" + qdb.sql_connection.TRN.add(sql, pt_ids) + else: + sql = """DELETE FROM qiita.parent_artifact + WHERE artifact_id IN %s""" + qdb.sql_connection.TRN.add(sql, [all_ids]) + + # Detach the artifacts from the study_artifact table + sql = "DELETE FROM qiita.study_artifact WHERE artifact_id IN %s" + qdb.sql_connection.TRN.add(sql, [all_ids]) + + # Detach the artifacts from the analysis_artifact table + sql = "DELETE FROM qiita.analysis_artifact WHERE artifact_id IN %s" + qdb.sql_connection.TRN.add(sql, [all_ids]) + + # Detach artifact from preparation_artifact + sql = """DELETE FROM qiita.preparation_artifact + WHERE artifact_id IN %s""" + qdb.sql_connection.TRN.add(sql, [all_ids]) + + # Delete the rows in the artifact table + sql = "DELETE FROM qiita.artifact WHERE artifact_id IN %s" + qdb.sql_connection.TRN.add(sql, [all_ids]) + + @classmethod + def archive(cls, artifact_id, clean_ancestors=True): + """Archive artifact with artifact_id + + Parameters + ---------- + artifact_id : int + The artifact to be archived + clean_ancestors : bool + If other childless artifacts should be deleted + + Raises + ------ + QiitaDBOperationNotPermittedError + If the artifact is not public + If the artifact_type is not BIOM + If the artifact belongs to an analysis + If the artifact has no parents (raw file) + """ + artifact = cls(artifact_id) + + if artifact.visibility != 'public': + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + 'Only public artifacts can be archived') + if artifact.artifact_type != 'BIOM': + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + 'Only BIOM artifacts can be archived') + if artifact.analysis is not None: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + 'Only non analysis artifacts can be archived') + if not artifact.parents: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + 'Only non raw artifacts can be archived') + + to_delete = [] + if clean_ancestors: + # let's find all ancestors that can be deleted (it has parents and + # no ancestors (that have no descendants), and delete them + to_delete = [x for x in artifact.ancestors.nodes() + if x.id != artifact_id and x.parents and + not [y for y in x.descendants.nodes() + if y.id not in (artifact_id, x.id)]] + # ignore artifacts that can and has been submitted to EBI + to_delete = [x for x in to_delete if not x.can_be_submitted_to_ebi + and not x.is_submitted_to_ebi + and not x.is_submitted_to_vamps] + + # get the log file so we can delete + fids = [x['fp_id'] for x in artifact.filepaths + if x['fp_type'] == 'log'] + + archive_data = dumps({"merging_scheme": artifact.merging_scheme}) + with qdb.sql_connection.TRN: + artifact._set_visibility('archived', propagate=False) + sql = 'DELETE FROM qiita.parent_artifact WHERE artifact_id = %s' + qdb.sql_connection.TRN.add(sql, [artifact_id]) + + sql = '''DELETE FROM qiita.artifact_output_processing_job + WHERE artifact_id = %s''' + qdb.sql_connection.TRN.add(sql, [artifact_id]) + + if fids: + sql = '''DELETE FROM qiita.artifact_filepath + WHERE filepath_id IN %s''' + qdb.sql_connection.TRN.add(sql, [tuple(fids)]) + + sql = """UPDATE qiita.{0} + SET archive_data = %s + WHERE artifact_id = %s""".format(cls._table) + qdb.sql_connection.TRN.add(sql, [archive_data, artifact_id]) + + qdb.sql_connection.TRN.execute() + + # cleaning the extra artifacts + for x in to_delete: + x._set_visibility('sandbox', propagate=False) + cls.delete(x.id) + + @property + def name(self): + """The name of the artifact + + Returns + ------- + str + The artifact name + """ + with qdb.sql_connection.TRN: + sql = """SELECT name + FROM qiita.artifact + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @name.setter + def name(self, value): + """Set the name of the artifact + + Parameters + ---------- + value : str + The new artifact's name + + Raises + ------ + ValueError + If `value` contains more than 35 chars + """ + sql = """UPDATE qiita.artifact + SET name = %s + WHERE artifact_id = %s""" + qdb.sql_connection.perform_as_transaction(sql, [value, self.id]) + + @property + def timestamp(self): + """The timestamp when the artifact was generated + + Returns + ------- + datetime + The timestamp when the artifact was generated + """ + with qdb.sql_connection.TRN: + sql = """SELECT generated_timestamp + FROM qiita.artifact + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def processing_parameters(self): + """The processing parameters used to generate the artifact + + Returns + ------- + qiita_db.software.Parameters or None + The parameters used to generate the artifact if it has parents. + None otherwise. + """ + with qdb.sql_connection.TRN: + sql = """SELECT command_id, command_parameters + FROM qiita.artifact + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + # Only one row will be returned + res = qdb.sql_connection.TRN.execute_fetchindex()[0] + if res[0] is None: + return None + return qdb.software.Parameters.load( + qdb.software.Command(res[0]), values_dict=res[1]) + + @property + def visibility(self): + """The visibility of the artifact + + Returns + ------- + str + The visibility of the artifact + """ + with qdb.sql_connection.TRN: + sql = """SELECT visibility + FROM qiita.artifact + JOIN qiita.visibility USING (visibility_id) + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + def _set_visibility(self, value, propagate=True): + "helper method to split validation and actual set of the visibility" + # In order to correctly propagate the visibility we need to find + # the root of this artifact and then propagate to all the artifacts + vis_id = qdb.util.convert_to_id(value, "visibility") + + if propagate: + sql = "SELECT * FROM qiita.find_artifact_roots(%s)" + qdb.sql_connection.TRN.add(sql, [self.id]) + root_id = qdb.sql_connection.TRN.execute_fetchlast() + root = qdb.artifact.Artifact(root_id) + # these are the ids of all the children from the root + ids = [a.id for a in root.descendants.nodes()] + else: + ids = [self.id] + + sql = """UPDATE qiita.artifact + SET visibility_id = %s + WHERE artifact_id IN %s""" + qdb.sql_connection.perform_as_transaction(sql, [vis_id, tuple(ids)]) + + @visibility.setter + def visibility(self, value): + """Sets the visibility of the artifact + + Parameters + ---------- + value : str + The new visibility of the artifact + + Notes + ----- + The visibility of an artifact is propagated to its ancestors, but it + only applies when the new visibility is more open than before. + """ + with qdb.sql_connection.TRN: + # first let's check that this is a valid visibility + study = self.study + + # then let's check that the sample/prep info files have the correct + # restrictions + if value != 'sandbox' and study is not None: + reply = study.sample_template.validate_restrictions() + success = [not reply[0]] + message = [reply[1]] + for pt in self.prep_templates: + reply = pt.validate_restrictions() + success.append(not reply[0]) + message.append(reply[1]) + if any(success): + raise ValueError( + "Errors in your info files:%s" % '\n'.join(message)) + + self._set_visibility(value) + + @property + def artifact_type(self): + """The artifact type + + Returns + ------- + str + The artifact type + """ + with qdb.sql_connection.TRN: + sql = """SELECT artifact_type + FROM qiita.artifact + JOIN qiita.artifact_type USING (artifact_type_id) + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def data_type(self): + """The data type of the artifact + + Returns + ------- + str + The artifact data type + """ + with qdb.sql_connection.TRN: + sql = """SELECT data_type + FROM qiita.artifact + JOIN qiita.data_type USING (data_type_id) + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def can_be_submitted_to_ebi(self): + """Whether the artifact can be submitted to EBI or not + + Returns + ------- + bool + True if the artifact can be submitted to EBI. False otherwise. + """ + with qdb.sql_connection.TRN: + # we should always return False if this artifact is not directly + # attached to the prep_template or is the second after. In other + # words has more that one processing step behind it + fine_to_send = [] + fine_to_send.extend([pt.artifact for pt in self.prep_templates]) + fine_to_send.extend([c for a in fine_to_send if a is not None + for c in a.children]) + if self not in fine_to_send: + return False + + sql = """SELECT can_be_submitted_to_ebi + FROM qiita.artifact_type + JOIN qiita.artifact USING (artifact_type_id) + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def is_submitted_to_ebi(self): + """Whether the artifact has been submitted to EBI or not + + Returns + ------- + bool + True if the artifact has been submitted to EBI. False otherwise + + Raises + ------ + QiitaDBOperationNotPermittedError + If the artifact cannot be submitted to EBI + """ + with qdb.sql_connection.TRN: + if not self.can_be_submitted_to_ebi: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + "Artifact %s cannot be submitted to EBI" % self.id) + sql = """SELECT EXISTS( + SELECT * + FROM qiita.ebi_run_accession + WHERE artifact_id = %s)""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def ebi_run_accessions(self): + """The EBI run accessions attached to this artifact + + Returns + ------- + dict of {str: str} + The EBI run accessions keyed by sample id + + Raises + ------ + QiitaDBOperationNotPermittedError + If the artifact cannot be submitted to EBI + """ + with qdb.sql_connection.TRN: + if not self.can_be_submitted_to_ebi: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + "Artifact %s cannot be submitted to EBI" % self.id) + sql = """SELECT sample_id, ebi_run_accession + FROM qiita.ebi_run_accession + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return {s_id: ebi_acc for s_id, ebi_acc in + qdb.sql_connection.TRN.execute_fetchindex()} + + @ebi_run_accessions.setter + def ebi_run_accessions(self, values): + """Set the EBI run accession attached to this artifact + + Parameters + ---------- + values : dict of {str: str} + The EBI accession number keyed by sample id + + Raises + ------ + QiitaDBOperationNotPermittedError + If the artifact cannot be submitted to EBI + If the artifact has been already submitted to EBI + """ + with qdb.sql_connection.TRN: + if not self.can_be_submitted_to_ebi: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + "Artifact %s cannot be submitted to EBI" % self.id) + + sql = """SELECT EXISTS(SELECT * + FROM qiita.ebi_run_accession + WHERE artifact_id = %s)""" + qdb.sql_connection.TRN.add(sql, [self.id]) + if qdb.sql_connection.TRN.execute_fetchlast(): + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + "Artifact %s already submitted to EBI" % self.id) + + sql = """INSERT INTO qiita.ebi_run_accession + (sample_id, artifact_id, ebi_run_accession) + VALUES (%s, %s, %s)""" + sql_args = [[sample, self.id, accession] + for sample, accession in values.items()] + qdb.sql_connection.TRN.add(sql, sql_args, many=True) + qdb.sql_connection.TRN.execute() + + @property + def can_be_submitted_to_vamps(self): + """Whether the artifact can be submitted to VAMPS or not + + Returns + ------- + bool + True if the artifact can be submitted to VAMPS. False otherwise. + """ + with qdb.sql_connection.TRN: + sql = """SELECT can_be_submitted_to_vamps + FROM qiita.artifact_type + JOIN qiita.artifact USING (artifact_type_id) + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def is_submitted_to_vamps(self): + """Whether if the artifact has been submitted to VAMPS or not + + Returns + ------- + bool + True if the artifact has been submitted to VAMPS. False otherwise + + Raises + ------ + QiitaDBOperationNotPermittedError + If the artifact cannot be submitted to VAMPS + """ + with qdb.sql_connection.TRN: + if not self.can_be_submitted_to_vamps: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + "Artifact %s cannot be submitted to VAMPS" % self.id) + sql = """SELECT submitted_to_vamps + FROM qiita.artifact + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @is_submitted_to_vamps.setter + def is_submitted_to_vamps(self, value): + """Set if the artifact has been submitted to VAMPS + + Parameters + ---------- + value : bool + Whether the artifact has been submitted to VAMPS or not + + Raises + ------ + QiitaDBOperationNotPermittedError + If the artifact cannot be submitted to VAMPS + """ + if not self.can_be_submitted_to_vamps: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + "Artifact %s cannot be submitted to VAMPS" % self.id) + sql = """UPDATE qiita.artifact + SET submitted_to_vamps = %s + WHERE artifact_id = %s""" + qdb.sql_connection.perform_as_transaction(sql, [value, self.id]) + + @property + def filepaths(self): + """Returns the filepaths associated with the artifact + + Returns + ------- + list of dict + A list of dict as defined by qiita_db.util.retrieve_filepaths + """ + return qdb.util.retrieve_filepaths( + "artifact_filepath", "artifact_id", self.id, sort='ascending') + + @property + def html_summary_fp(self): + """Returns the HTML summary filepath + + Returns + ------- + tuple of (int, str) + The filepath id and the path to the HTML summary + """ + fps = qdb.util.retrieve_filepaths("artifact_filepath", "artifact_id", + self.id, fp_type='html_summary') + if fps: + # If fps is not the empty list, then we have exactly one file + # retrieve_filepaths returns a list of lists of 3 values: the + # filepath id, the filepath and the filepath type. We don't want + # to return the filepath type here, so just grabbing the first and + # second element of the list + res = (fps[0]['fp_id'], fps[0]['fp']) + else: + res = None + + return res + + def set_html_summary(self, html_fp, support_dir=None): + """Sets the HTML summary of the artifact + + Parameters + ---------- + html_fp : str + Path to the new HTML summary + support_dir : str + Path to the directory containing any support files needed by + the HTML file + """ + with qdb.sql_connection.TRN: + old_summs = self.html_summary_fp + to_delete_fps = [] + if old_summs: + # Delete from the DB current HTML summary; below we will remove + # files, if necessary + to_delete_ids = [] + for x in self.filepaths: + if x['fp_type'] in ('html_summary', 'html_summary_dir'): + to_delete_ids.append([x['fp_id']]) + to_delete_fps.append(x['fp']) + # From the artifact_filepath table + sql = """DELETE FROM qiita.artifact_filepath + WHERE filepath_id = %s""" + qdb.sql_connection.TRN.add(sql, to_delete_ids, many=True) + # From the filepath table + sql = "DELETE FROM qiita.filepath WHERE filepath_id=%s" + qdb.sql_connection.TRN.add(sql, to_delete_ids, many=True) + + # Add the new HTML summary + filepaths = [(html_fp, 'html_summary')] + if support_dir is not None: + filepaths.append((support_dir, 'html_summary_dir')) + fp_ids = qdb.util.insert_filepaths( + filepaths, self.id, self.artifact_type) + sql = """INSERT INTO qiita.artifact_filepath + (artifact_id, filepath_id) + VALUES (%s, %s)""" + sql_args = [[self.id, id_] for id_ in fp_ids] + qdb.sql_connection.TRN.add(sql, sql_args, many=True) + qdb.sql_connection.TRN.execute() + + # to avoid deleting potentially necessary files, we are going to add + # that check after the previous transaction is commited + if to_delete_fps: + for x in self.filepaths: + if x['fp'] in to_delete_fps: + to_delete_fps.remove(x['fp']) + + for fp in to_delete_fps: + if isfile(fp): + remove(fp) + else: + rmtree(fp) + + @property + def parents(self): + """Returns the parents of the artifact + + Returns + ------- + list of qiita_db.artifact.Artifact + The parent artifacts + """ + with qdb.sql_connection.TRN: + sql = """SELECT parent_id + FROM qiita.parent_artifact + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return [Artifact(p_id) + for p_id in qdb.sql_connection.TRN.execute_fetchflatten()] + + def _create_lineage_graph_from_edge_list(self, edge_list): + """Generates an artifact graph from the given `edge_list` + + Parameters + ---------- + edge_list : list of (int, int) + List of (parent_artifact_id, artifact_id) + + Returns + ------- + networkx.DiGraph + The graph representing the artifact lineage stored in `edge_list` + """ + lineage = nx.DiGraph() + # In case the edge list is empty, only 'self' is present in the graph + if edge_list: + # By creating all the artifacts here we are saving DB calls + nodes = {a_id: Artifact(a_id) + for a_id in set(chain.from_iterable(edge_list))} + + for parent, child in edge_list: + lineage.add_edge(nodes[parent], nodes[child]) + else: + lineage.add_node(self) + + return lineage + + @property + def ancestors(self): + """Returns the ancestors of the artifact + + Returns + ------- + networkx.DiGraph + The ancestors of the artifact + """ + with qdb.sql_connection.TRN: + sql = """SELECT parent_id, artifact_id + FROM qiita.artifact_ancestry(%s)""" + qdb.sql_connection.TRN.add(sql, [self.id]) + edges = qdb.sql_connection.TRN.execute_fetchindex() + return self._create_lineage_graph_from_edge_list(edges) + + @property + def descendants(self): + """Returns the descendants of the artifact + + Returns + ------- + networkx.DiGraph + The descendants of the artifact + """ + with qdb.sql_connection.TRN: + sql = """SELECT parent_id, artifact_id + FROM qiita.artifact_descendants(%s)""" + qdb.sql_connection.TRN.add(sql, [self.id]) + edges = qdb.sql_connection.TRN.execute_fetchindex() + return self._create_lineage_graph_from_edge_list(edges) + + @property + def descendants_with_jobs(self): + """Returns the descendants of the artifact with their jobs + + Returns + ------- + networkx.DiGraph + The descendants of the artifact + """ + def _add_edge(edges, src, dest): + """Aux function to add the edge (src, dest) to edges""" + edge = (src, dest) + if edge not in edges: + edges.add(edge) + + with qdb.sql_connection.TRN: + sql = """SELECT processing_job_id, input_id, output_id + FROM qiita.artifact_descendants_with_jobs(%s)""" + qdb.sql_connection.TRN.add(sql, [self.id]) + sql_edges = qdb.sql_connection.TRN.execute_fetchindex() + + # helper function to reduce code duplication + def _helper(sql_edges, edges, nodes): + for jid, pid, cid in sql_edges: + if jid not in nodes: + nodes[jid] = ('job', + qdb.processing_job.ProcessingJob(jid)) + if pid not in nodes: + nodes[pid] = ('artifact', qdb.artifact.Artifact(pid)) + if cid not in nodes: + nodes[cid] = ('artifact', qdb.artifact.Artifact(cid)) + edges.add((nodes[pid], nodes[jid])) + edges.add((nodes[jid], nodes[cid])) + + lineage = nx.DiGraph() + edges = set() + nodes = dict() + extra_edges = set() + extra_nodes = dict() + if sql_edges: + _helper(sql_edges, edges, nodes) + else: + nodes[self.id] = ('artifact', self) + lineage.add_node(nodes[self.id]) + # if this is an Analysis we need to check if there are extra + # edges/nodes as there is a chance that there are connecions + # between them + if self.analysis is not None: + roots = [a for a in self.analysis.artifacts + if not a.parents and a != self] + for r in roots: + # add the root to the options then their children + extra_nodes[r.id] = ('artifact', r) + qdb.sql_connection.TRN.add(sql, [r.id]) + sql_edges = qdb.sql_connection.TRN.execute_fetchindex() + _helper(sql_edges, extra_edges, extra_nodes) + + # The code above returns all the jobs that have been successfully + # executed. We need to add all the jobs that are in all the other + # status. Approach: Loop over all the artifacts and add all the + # jobs that have been attached to them. + visited = set() + queue = list(nodes.keys()) + while queue: + current = queue.pop(0) + if current not in visited: + visited.add(current) + n_type, n_obj = nodes[current] + if n_type == 'artifact': + # Add all the jobs to the queue + for job in n_obj.jobs(): + queue.append(job.id) + if job.id not in nodes: + nodes[job.id] = ('job', job) + + elif n_type == 'job': + # skip private and artifact definition jobs as they + # don't create new artifacts and they would create + # edges without artifacts + they can be safely ignored + if n_obj.command.software.type in { + 'private', 'artifact definition'}: + continue + jstatus = n_obj.status + # If the job is in success we don't need to do anything + # else since it would've been added by the code above + if jstatus != 'success': + + if jstatus != 'error': + # If the job is not errored, we can add the + # future outputs and the children jobs to + # the graph. + + # Add all the job outputs as new nodes + for o_name, o_type in n_obj.command.outputs: + node_id = '%s:%s' % (n_obj.id, o_name) + node = TypeNode( + id=node_id, job_id=n_obj.id, + name=o_name, type=o_type) + queue.append(node_id) + if node_id not in nodes: + nodes[node_id] = ('type', node) + + # Add all his children jobs to the queue + for cjob in n_obj.children: + queue.append(cjob.id) + if cjob.id not in nodes: + nodes[cjob.id] = ('job', cjob) + + # including the outputs + for o_name, o_type in cjob.command.outputs: + node_id = '%s:%s' % (cjob.id, o_name) + node = TypeNode( + id=node_id, job_id=cjob.id, + name=o_name, type=o_type) + if node_id not in nodes: + nodes[node_id] = ('type', node) + + # Connect the job with his input artifacts, the + # input artifacts may or may not exist yet, so we + # need to check both the input_artifacts and the + # pending properties + for in_art in n_obj.input_artifacts: + iid = in_art.id + if iid not in nodes and iid in extra_nodes: + nodes[iid] = extra_nodes[iid] + _add_edge(edges, nodes[iid], nodes[n_obj.id]) + + pending = n_obj.pending + for pred_id in pending: + for pname in pending[pred_id]: + in_node_id = '%s:%s' % ( + pred_id, pending[pred_id][pname]) + _add_edge(edges, nodes[in_node_id], + nodes[n_obj.id]) + + elif n_type == 'type': + # Connect this 'future artifact' with the job that will + # generate it + _add_edge(edges, nodes[n_obj.job_id], nodes[current]) + else: + raise ValueError('Unrecognized type: %s' % n_type) + + # Add all edges to the lineage graph - adding the edges creates the + # nodes in networkx + for source, dest in edges: + lineage.add_edge(source, dest) + + return lineage + + @property + def children(self): + """Returns the list of children of the artifact + + Returns + ------- + list of qiita_db.artifact.Artifact + The children artifacts + """ + with qdb.sql_connection.TRN: + sql = """SELECT artifact_id + FROM qiita.parent_artifact + WHERE parent_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return [Artifact(c_id) + for c_id in qdb.sql_connection.TRN.execute_fetchflatten()] + + @property + def youngest_artifact(self): + """Returns the youngest artifact of the artifact's lineage + + Returns + ------- + qiita_db.artifact.Artifact + The youngest descendant of the artifact's lineage + """ + with qdb.sql_connection.TRN: + sql = """SELECT artifact_id + FROM qiita.artifact_descendants(%s) + JOIN qiita.artifact USING (artifact_id) + WHERE visibility_id NOT IN %s + ORDER BY generated_timestamp DESC + LIMIT 1""" + qdb.sql_connection.TRN.add( + sql, [self.id, qdb.util.artifact_visibilities_to_skip()]) + a_id = qdb.sql_connection.TRN.execute_fetchindex() + # If the current artifact has no children, the previous call will + # return an empty list, so the youngest artifact in the lineage is + # the current artifact. On the other hand, if it has descendants, + # the id of the youngest artifact will be in a_id[0][0] + result = Artifact(a_id[0][0]) if a_id else self + + return result + + @property + def prep_templates(self): + """The prep templates attached to this artifact + + Returns + ------- + list of qiita_db.metadata_template.PrepTemplate + """ + with qdb.sql_connection.TRN: + sql = """SELECT prep_template_id + FROM qiita.preparation_artifact + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return [qdb.metadata_template.prep_template.PrepTemplate(pt_id) + for pt_id in qdb.sql_connection.TRN.execute_fetchflatten()] + + @property + def study(self): + """The study to which the artifact belongs to + + Returns + ------- + qiita_db.study.Study or None + The study that owns the artifact, if any + """ + with qdb.sql_connection.TRN: + sql = """SELECT study_id + FROM qiita.study_artifact + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + res = qdb.sql_connection.TRN.execute_fetchindex() + return qdb.study.Study(res[0][0]) if res else None + + @property + def analysis(self): + """The analysis to which the artifact belongs to + + Returns + ------- + qiita_db.analysis.Analysis or None + The analysis that owns the artifact, if any + """ + with qdb.sql_connection.TRN: + sql = """SELECT analysis_id + FROM qiita.analysis_artifact + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + res = qdb.sql_connection.TRN.execute_fetchindex() + return qdb.analysis.Analysis(res[0][0]) if res else None + + @property + def merging_scheme(self): + """The merging scheme of this artifact_type + + Returns + ------- + str, str + The human readable merging scheme and the parent software + information for this artifact + """ + vid = qdb.util.convert_to_id(self.visibility, "visibility") + if vid in qdb.util.artifact_visibilities_to_skip(): + with qdb.sql_connection.TRN: + sql = f"""SELECT archive_data + FROM qiita.{self._table} + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + archive_data = qdb.sql_connection.TRN.execute_fetchlast() + merging_schemes = [archive_data['merging_scheme'][0]] + parent_softwares = [archive_data['merging_scheme'][1]] + else: + processing_params = self.processing_parameters + if processing_params is None: + return '', '' + + cmd_name = processing_params.command.name + ms = processing_params.command.merging_scheme + afps = [x['fp'] for x in self.filepaths + if x['fp'].endswith('biom')] + + merging_schemes = [] + parent_softwares = [] + # this loop is necessary as in theory an artifact can be + # generated from multiple prep info files + for p in self.parents: + pparent = p.processing_parameters + # if parent is None, then is a direct upload; for example + # per_sample_FASTQ in shotgun data + if pparent is None: + parent_cmd_name = None + parent_merging_scheme = None + parent_pp = None + parent_software = 'N/A' + else: + parent_cmd_name = pparent.command.name + parent_merging_scheme = pparent.command.merging_scheme + parent_pp = pparent.values + psoftware = pparent.command.software + parent_software = '%s v%s' % ( + psoftware.name, psoftware.version) + + merging_schemes.append(qdb.util.human_merging_scheme( + cmd_name, ms, parent_cmd_name, parent_merging_scheme, + processing_params.values, afps, parent_pp)) + parent_softwares.append(parent_software) + + return ', '.join(merging_schemes), ', '.join(parent_softwares) + + @property + def being_deleted_by(self): + """The running job that is deleting this artifact + + Returns + ------- + qiita_db.processing_job.ProcessingJob + The running job that is deleting this artifact, None if it + doesn't exist + """ + + with qdb.sql_connection.TRN: + sql = """ + SELECT processing_job_id FROM qiita.artifact_processing_job + LEFT JOIN qiita.processing_job using (processing_job_id) + LEFT JOIN qiita.processing_job_status using ( + processing_job_status_id) + LEFT JOIN qiita.software_command using (command_id) + WHERE artifact_id = %s AND name = 'delete_artifact' AND + processing_job_status in ( + 'running', 'queued', 'in_construction')""" + qdb.sql_connection.TRN.add(sql, [self.id]) + res = qdb.sql_connection.TRN.execute_fetchindex() + return qdb.processing_job.ProcessingJob(res[0][0]) if res else None + + @property + def has_human(self): + has_human = False + # we are going to check the metadata if: + # - the prep data_type is _not_ target gene + # - the prep is not current_human_filtering + # - if the artifact_type is 'per_sample_FASTQ' + pts = self.prep_templates + tgs = qdb.metadata_template.constants.TARGET_GENE_DATA_TYPES + ntg = any([pt.data_type() not in tgs for pt in pts]) + chf = any([not pt.current_human_filtering for pt in pts]) + if ntg and chf and self.artifact_type == 'per_sample_FASTQ': + st = self.study.sample_template + if 'env_package' in st.categories: + sql = f"""SELECT DISTINCT sample_values->>'env_package' + FROM qiita.sample_{st.id} WHERE sample_id in ( + SELECT sample_id from qiita.preparation_artifact + LEFT JOIN qiita.prep_template_sample USING ( + prep_template_id) + WHERE artifact_id = {self.id})""" + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add(sql) + for v in qdb.sql_connection.TRN.execute_fetchflatten(): + # str is needed as v could be None + if str(v).startswith('human-'): + has_human = True + break + + return has_human + + def jobs(self, cmd=None, status=None, show_hidden=False): + """Jobs that used this artifact as input + + Parameters + ---------- + cmd : qiita_db.software.Command, optional + If provided, only jobs that executed this command will be returned + status : str, optional + If provided, only jobs in this status will be returned + show_hidden : bool, optional + If true, return also the "hidden" jobs + + Returns + ------- + list of qiita_db.processing_job.ProcessingJob + The list of jobs that used this artifact as input + """ + with qdb.sql_connection.TRN: + sql = """SELECT processing_job_id + FROM qiita.artifact_processing_job + JOIN qiita.processing_job USING (processing_job_id) + JOIN qiita.processing_job_status + USING (processing_job_status_id) + WHERE artifact_id = %s""" + sql_args = [self.id] + + if cmd: + sql = "{} AND command_id = %s".format(sql) + sql_args.append(cmd.id) + + if status: + sql = "{} AND processing_job_status = %s".format(sql) + sql_args.append(status) + + if not show_hidden: + sql = "{} AND hidden = %s".format(sql) + sql_args.append(False) + + qdb.sql_connection.TRN.add(sql, sql_args) + return [qdb.processing_job.ProcessingJob(jid) + for jid in qdb.sql_connection.TRN.execute_fetchflatten()] + + @property + def get_commands(self): + """Returns the active commands that can process this kind of artifact + + Returns + ------- + list of qiita_db.software.Command + The commands that can process the given artifact tyoes + """ + dws = [] + with qdb.sql_connection.TRN: + # get all the possible commands + sql = """SELECT DISTINCT qiita.command_parameter.command_id + FROM qiita.artifact + JOIN qiita.parameter_artifact_type + USING (artifact_type_id) + JOIN qiita.command_parameter USING (command_parameter_id) + JOIN qiita.software_command ON ( + qiita.command_parameter.command_id = + qiita.software_command.command_id) + WHERE artifact_id = %s AND active = True""" + if self.analysis is None: + sql += " AND is_analysis = False" + # get the workflows that match this artifact so we can filter + # the available commands based on the commands in the worflows + # for that artifact - except is the artifact_type == 'BIOM' + if self.artifact_type != 'BIOM': + dws = [w for w in qdb.software.DefaultWorkflow.iter() + if self.data_type in w.data_type] + else: + sql += " AND is_analysis = True" + + qdb.sql_connection.TRN.add(sql, [self.id]) + cids = set(qdb.sql_connection.TRN.execute_fetchflatten()) + + if dws: + cmds = {n.default_parameter.command.id + for w in dws for n in w.graph.nodes} + cids = cmds & cids + + return [qdb.software.Command(cid) for cid in cids] + + @property + def human_reads_filter_method(self): + """The human_reads_filter_method of the artifact + + Returns + ------- + str + The human_reads_filter_method name + """ + with qdb.sql_connection.TRN: + sql = """SELECT human_reads_filter_method + FROM qiita.artifact + LEFT JOIN qiita.human_reads_filter_method + USING (human_reads_filter_method_id) + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @human_reads_filter_method.setter + def human_reads_filter_method(self, value): + """Set the human_reads_filter_method of the artifact + + Parameters + ---------- + value : str + The new artifact's human_reads_filter_method + + Raises + ------ + ValueError + If `value` doesn't exist in the database + """ + with qdb.sql_connection.TRN: + sql = """SELECT human_reads_filter_method_id + FROM qiita.human_reads_filter_method + WHERE human_reads_filter_method = %s""" + qdb.sql_connection.TRN.add(sql, [value]) + idx = qdb.sql_connection.TRN.execute_fetchflatten() + + if len(idx) == 0: + raise ValueError( + f'"{value}" is not a valid human_reads_filter_method') + + sql = """UPDATE qiita.artifact + SET human_reads_filter_method_id = %s + WHERE artifact_id = %s""" + qdb.sql_connection.TRN.add(sql, [idx[0], self.id])