Datasets
Models
Diff of
/qiita_db/software.py
[000000]
..
[879b32]
Switch to side-by-side view
--- a +++ b/qiita_db/software.py @@ -0,0 +1,2060 @@ +# ----------------------------------------------------------------------------- +# 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 json import dumps, loads +from copy import deepcopy +import inspect +import warnings + +import networkx as nx + +from qiita_core.qiita_settings import qiita_config +import qiita_db as qdb + +from configparser import ConfigParser + + +class Command(qdb.base.QiitaObject): + r"""An executable command available in the system + + Attributes + ---------- + active + post_processing_cmd + analysis_only + default_parameter_sets + description + merging_scheme + name + naming_order + optional_parameters + outputs + parameters + required_parameters + software + description + cli + parameters_table + + Methods + ------- + _check_id + activate + + Class Methods + ------------- + create + exists + get_commands_by_input_type + get_html_generator + get_validator + + See Also + -------- + qiita_db.software.Software + """ + _table = "software_command" + + @classmethod + def get_commands_by_input_type(cls, artifact_types, active_only=True, + exclude_analysis=True, prep_type=None): + """Returns the commands that can process the given artifact types + + Parameters + ---------- + artifact_type : list of str + The artifact types + active_only : bool, optional + If True, return only active commands, otherwise return all commands + Default: True + exclude_analysis : bool, optional + If True, return commands that are not part of the analysis pipeline + + Returns + ------- + generator of qiita_db.software.Command + The commands that can process the given artifact tyoes + """ + with qdb.sql_connection.TRN: + sql = """SELECT DISTINCT command_id + FROM qiita.command_parameter + JOIN qiita.parameter_artifact_type + USING (command_parameter_id) + JOIN qiita.artifact_type USING (artifact_type_id) + JOIN qiita.software_command USING (command_id) + WHERE artifact_type IN %s""" + if active_only: + sql += " AND active = True" + if exclude_analysis: + sql += " AND is_analysis = False" + qdb.sql_connection.TRN.add(sql, [tuple(artifact_types)]) + cids = set(qdb.sql_connection.TRN.execute_fetchflatten()) + + if prep_type is not None: + dws = [w for w in qdb.software.DefaultWorkflow.iter() + if prep_type in w.data_type] + if dws: + cmds = {n.default_parameter.command.id + for w in dws for n in w.graph.nodes} + cids = cmds & cids + + return [cls(cid) for cid in cids] + + @classmethod + def get_html_generator(cls, artifact_type): + """Returns the command that generete the HTML for the given artifact + + Parameters + ---------- + artifact_type : str + The artifact type to search the HTML generator for + + Returns + ------- + qiita_db.software.Command + The newly created command + + Raises + ------ + qdb.exceptions.QiitaDBError when the generete the HTML command can't + be found + """ + with qdb.sql_connection.TRN: + sql = """SELECT command_id + FROM qiita.software_command + JOIN qiita.software_artifact_type USING (software_id) + JOIN qiita.artifact_type USING (artifact_type_id) + WHERE artifact_type = %s + AND name = 'Generate HTML summary' + AND active = true""" + qdb.sql_connection.TRN.add(sql, [artifact_type]) + try: + res = qdb.sql_connection.TRN.execute_fetchlast() + except IndexError: + raise qdb.exceptions.QiitaDBError( + "There is no command to generate the HTML summary for " + "artifact type '%s'" % artifact_type) + + return cls(res) + + @classmethod + def get_validator(cls, artifact_type): + """Returns the command that validates the given artifact + + Parameters + ---------- + artifact_type : str + The artifact type to search the Validate for + + Returns + ------- + qiita_db.software.Command + The newly created command + + Raises + ------ + qdb.exceptions.QiitaDBError when the Validate command can't be found + """ + with qdb.sql_connection.TRN: + sql = """SELECT command_id + FROM qiita.software_command + JOIN qiita.software_artifact_type USING (software_id) + JOIN qiita.artifact_type USING (artifact_type_id) + WHERE artifact_type = %s + AND name = 'Validate' + AND active = true""" + qdb.sql_connection.TRN.add(sql, [artifact_type]) + try: + res = qdb.sql_connection.TRN.execute_fetchlast() + except IndexError: + raise qdb.exceptions.QiitaDBError( + "There is no command to generate the Validate for " + "artifact type '%s'" % artifact_type) + + return cls(res) + + def _check_id(self, id_): + """Check that the provided ID actually exists in the database + + Parameters + ---------- + id_ : int + The ID to test + + Notes + ----- + This function overwrites the base function, as the sql layout doesn't + follow the same conventions done in the other classes. + """ + with qdb.sql_connection.TRN: + sql = """SELECT EXISTS( + SELECT * + FROM qiita.software_command + WHERE command_id = %s)""" + qdb.sql_connection.TRN.add(sql, [id_]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @classmethod + def exists(cls, software, name): + """Checks if the command already exists in the system + + Parameters + ---------- + qiita_db.software.Software + The software to which this command belongs to. + name : str + The name of the command + + Returns + ------- + bool + Whether the command exists in the system or not + """ + with qdb.sql_connection.TRN: + sql = """SELECT EXISTS(SELECT * + FROM qiita.software_command + WHERE software_id = %s + AND name = %s)""" + qdb.sql_connection.TRN.add(sql, [software.id, name]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @classmethod + def create(cls, software, name, description, parameters, outputs=None, + analysis_only=False): + r"""Creates a new command in the system + + The supported types for the parameters are: + - string: the parameter is a free text input + - integer: the parameter is an integer + - float: the parameter is a float + - artifact: the parameter is an artifact instance, the artifact id + will be stored + - reference: the parameter is a reference instance, the reference + id will be stored + - choice: the format of this should be `choice:<json-dump-of-list>` + in which json-dump-of-list is the JSON dump of a list containing + the acceptable values + + Parameters + ---------- + software : qiita_db.software.Software + The software to which this command belongs to. + name : str + The name of the command + description : str + The description of the command + parameters : dict + The description of the parameters that this command received. The + format is: {parameter_name: (parameter_type, default, name_order, + check_biom_merge, qiita_optional_parameter (optional))}, + where parameter_name, parameter_type and default are strings, + name_order is an optional integer value and check_biom_merge is + an optional boolean value. name_order is used to specify the order + of the parameter when automatically naming the artifacts. + check_biom_merge is used when merging artifacts in the analysis + pipeline. qiita_optional_parameter is an optional bool to "force" + the parameter to be optional + outputs : dict, optional + The description of the outputs that this command generated. The + format is either {output_name: artifact_type} or + {output_name: (artifact_type, check_biom_merge)} + analysis_only : bool, optional + If true, then the command will only be available on the analysis + pipeline. Default: False. + + Returns + ------- + qiita_db.software.Command + The newly created command + + Raises + ------ + QiitaDBError + - If parameters is empty + - If the parameters dictionary is malformed + - If one of the parameter types is not supported + - If the default value of a choice parameter is not listed in + the available choices + QiitaDBDuplicateError + - If the command already exists + + Notes + ----- + If the default value for a parameter is NULL, then the parameter will + be required. On the other hand, if it is provided, the parameter will + be optional and the default value will be used when the user doesn't + overwrite it. + """ + # Perform some sanity checks in the parameters dictionary + if not parameters: + raise qdb.exceptions.QiitaDBError( + "Error creating command %s. At least one parameter should " + "be provided." % name) + sql_param_values = [] + sql_artifact_params = [] + for pname, vals in parameters.items(): + qiita_optional_parameter = False + if 'qiita_optional_parameter' in vals: + qiita_optional_parameter = True + vals.remove('qiita_optional_parameter') + lenvals = len(vals) + if lenvals == 2: + ptype, dflt = vals + name_order = None + check_biom_merge = False + elif lenvals == 4: + ptype, dflt, name_order, check_biom_merge = vals + else: + raise qdb.exceptions.QiitaDBError( + "Malformed parameters dictionary, the format should be " + "either {param_name: [parameter_type, default]} or " + "{parameter_name: (parameter_type, default, name_order, " + "check_biom_merge)}. Found: %s for parameter name %s" + % (vals, pname)) + + # Check that the type is one of the supported types + supported_types = ['string', 'integer', 'float', 'reference', + 'boolean', 'prep_template', 'analysis'] + if ptype not in supported_types and not ptype.startswith( + ('choice', 'mchoice', 'artifact')): + supported_types.extend(['choice', 'mchoice', 'artifact']) + raise qdb.exceptions.QiitaDBError( + "Unsupported parameters type '%s' for parameter %s. " + "Supported types are: %s" + % (ptype, pname, ', '.join(supported_types))) + + if ptype.startswith(('choice', 'mchoice')) and dflt is not None: + choices = set(loads(ptype.split(':')[1])) + dflt_val = dflt + if ptype.startswith('choice'): + # In the choice case, the dflt value is a single string, + # create a list with it the string on it to use the + # issuperset call below + dflt_val = [dflt_val] + else: + # jsonize the list to store it in the DB + dflt = dumps(dflt) + if not choices.issuperset(dflt_val): + raise qdb.exceptions.QiitaDBError( + "The default value '%s' for the parameter %s is not " + "listed in the available choices: %s" + % (dflt, pname, ', '.join(choices))) + + if ptype.startswith('artifact'): + atypes = loads(ptype.split(':')[1]) + sql_artifact_params.append( + [pname, 'artifact', atypes]) + else: + # a parameter will be required (not optional) if + # qiita_optional_parameter is false and there is the default + # value (dflt) is None + required = not qiita_optional_parameter and dflt is None + sql_param_values.append([pname, ptype, required, dflt, + name_order, check_biom_merge]) + + with qdb.sql_connection.TRN: + if cls.exists(software, name): + raise qdb.exceptions.QiitaDBDuplicateError( + "command", "software: %d, name: %s" + % (software.id, name)) + # Add the command to the DB + sql = """INSERT INTO qiita.software_command + (name, software_id, description, is_analysis) + VALUES (%s, %s, %s, %s) + RETURNING command_id""" + sql_params = [name, software.id, description, analysis_only] + qdb.sql_connection.TRN.add(sql, sql_params) + c_id = qdb.sql_connection.TRN.execute_fetchlast() + + # Add the parameters to the DB + sql = """INSERT INTO qiita.command_parameter + (command_id, parameter_name, parameter_type, + required, default_value, name_order, check_biom_merge) + VALUES (%s, %s, %s, %s, %s, %s, %s) + RETURNING command_parameter_id""" + sql_params = [ + [c_id, pname, p_type, reqd, default, no, chm] + for pname, p_type, reqd, default, no, chm in sql_param_values] + qdb.sql_connection.TRN.add(sql, sql_params, many=True) + qdb.sql_connection.TRN.execute() + + # Add the artifact parameters + sql_type = """INSERT INTO qiita.parameter_artifact_type + (command_parameter_id, artifact_type_id) + VALUES (%s, %s)""" + supported_types = [] + for pname, p_type, atypes in sql_artifact_params: + sql_params = [c_id, pname, p_type, True, None, None, False] + qdb.sql_connection.TRN.add(sql, sql_params) + pid = qdb.sql_connection.TRN.execute_fetchlast() + sql_params = [ + [pid, qdb.util.convert_to_id(at, 'artifact_type')] + for at in atypes] + qdb.sql_connection.TRN.add(sql_type, sql_params, many=True) + supported_types.extend([atid for _, atid in sql_params]) + + # If the software type is 'artifact definition', there are a couple + # of extra steps + if software.type == 'artifact definition': + # If supported types is not empty, link the software with these + # types + if supported_types: + sql = """INSERT INTO qiita.software_artifact_type + (software_id, artifact_type_id) + VALUES (%s, %s)""" + sql_params = [[software.id, atid] + for atid in supported_types] + qdb.sql_connection.TRN.add(sql, sql_params, many=True) + # If this is the validate command, we need to add the + # provenance and name parameters. These are used internally, + # that's why we are adding them here + if name == 'Validate': + sql = """INSERT INTO qiita.command_parameter + (command_id, parameter_name, parameter_type, + required, default_value) + VALUES (%s, 'name', 'string', 'False', + 'dflt_name'), + (%s, 'provenance', 'string', 'False', NULL) + """ + qdb.sql_connection.TRN.add(sql, [c_id, c_id]) + + # Add the outputs to the command + if outputs: + sql_args = [] + for pname, at in outputs.items(): + if isinstance(at, tuple): + sql_args.append( + [pname, c_id, + qdb.util.convert_to_id(at[0], 'artifact_type'), + at[1]]) + else: + try: + at_id = qdb.util.convert_to_id(at, 'artifact_type') + except qdb.exceptions.QiitaDBLookupError: + msg = (f'Error creating {software.name}, {name}, ' + f'{description} - Unknown artifact_type: ' + f'{at}') + raise ValueError(msg) + sql_args.append([pname, c_id, at_id, False]) + + sql = """INSERT INTO qiita.command_output + (name, command_id, artifact_type_id, + check_biom_merge) + VALUES (%s, %s, %s, %s)""" + qdb.sql_connection.TRN.add(sql, sql_args, many=True) + qdb.sql_connection.TRN.execute() + + return cls(c_id) + + @property + def software(self): + """The software to which this command belongs to + + Returns + ------- + qiita_db.software.Software + the software to which this command belongs to + """ + with qdb.sql_connection.TRN: + sql = """SELECT software_id + FROM qiita.software_command + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return Software(qdb.sql_connection.TRN.execute_fetchlast()) + + @property + def name(self): + """The name of the command + + Returns + ------- + str + The name of the command + """ + with qdb.sql_connection.TRN: + sql = """SELECT name + FROM qiita.software_command + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def post_processing_cmd(self): + """Additional processing commands required for merging + + Returns + ------- + str + Returns the additional processing command for merging + """ + with qdb.sql_connection.TRN: + sql = """SELECT post_processing_cmd + FROM qiita.software_command + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + + cmd = qdb.sql_connection.TRN.execute_fetchlast() + if cmd: + # assume correctly formatted json data + # load data into dictionary; don't return JSON + return loads(qdb.sql_connection.TRN.execute_fetchlast()) + + return None + + @property + def description(self): + """The description of the command + + Returns + ------- + str + The description of the command + """ + with qdb.sql_connection.TRN: + sql = """SELECT description + FROM qiita.software_command + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def parameters(self): + """Returns the parameters that the command accepts + + Returns + ------- + dict + Dictionary of {parameter_name: [ptype, dflt]} + """ + with qdb.sql_connection.TRN: + sql = """SELECT parameter_name, parameter_type, default_value + FROM qiita.command_parameter + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + res = qdb.sql_connection.TRN.execute_fetchindex() + return {pname: [ptype, dflt] for pname, ptype, dflt in res} + + @property + def required_parameters(self): + """Returns the required parameters that the command accepts + + Returns + ------- + dict + Dictionary of {parameter_name: ptype} + """ + with qdb.sql_connection.TRN: + sql = """SELECT command_parameter_id, parameter_name, + parameter_type, array_agg( + artifact_type ORDER BY artifact_type) AS + artifact_type + FROM qiita.command_parameter + LEFT JOIN qiita.parameter_artifact_type + USING (command_parameter_id) + LEFT JOIN qiita.artifact_type USING (artifact_type_id) + WHERE command_id = %s AND required = True + GROUP BY command_parameter_id""" + qdb.sql_connection.TRN.add(sql, [self.id]) + res = qdb.sql_connection.TRN.execute_fetchindex() + return {pname: (ptype, atype) for _, pname, ptype, atype in res} + + @property + def optional_parameters(self): + """Returns the optional parameters that the command accepts + + Returns + ------- + dict + Dictionary of {parameter_name: [ptype, default]} + """ + with qdb.sql_connection.TRN: + sql = """SELECT parameter_name, parameter_type, default_value + FROM qiita.command_parameter + WHERE command_id = %s AND required = false""" + qdb.sql_connection.TRN.add(sql, [self.id]) + res = qdb.sql_connection.TRN.execute_fetchindex() + + # Define a function to load the json storing the default parameters + # if ptype is multiple choice. When I added it to the for loop as + # a one liner if, made the code a bit hard to read + def dflt_fmt(dflt, ptype): + if ptype.startswith('mchoice'): + return loads(dflt) + return dflt + + return {pname: [ptype, dflt_fmt(dflt, ptype)] + for pname, ptype, dflt in res} + + @property + def default_parameter_sets(self): + """Returns the list of default parameter sets + + Returns + ------- + generator + generator of qiita_db.software.DefaultParameters + """ + with qdb.sql_connection.TRN: + sql = """SELECT default_parameter_set_id + FROM qiita.default_parameter_set + WHERE command_id = %s + ORDER BY default_parameter_set_id""" + qdb.sql_connection.TRN.add(sql, [self.id]) + res = qdb.sql_connection.TRN.execute_fetchflatten() + for pid in res: + yield DefaultParameters(pid) + + @property + def outputs(self): + """Returns the list of output artifact types + + Returns + ------- + list of str + The output artifact types + """ + with qdb.sql_connection.TRN: + sql = """SELECT name, artifact_type + FROM qiita.command_output + JOIN qiita.artifact_type USING (artifact_type_id) + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchindex() + + @property + def active(self): + """Returns if the command is active or not + + Returns + ------- + bool + Whether the command is active or not + + Notes + ----- + This method differentiates between commands based on analysis_only or + the software type. The commands that are not for analysis (processing) + and are from an artifact definition software will return as active + if they have the same name than a command that is active; this helps + for situations where the processing plugins are updated but some + commands didn't change its version. + """ + with qdb.sql_connection.TRN: + cmd_type = self.software.type + if self.analysis_only or cmd_type == 'artifact definition': + sql = """SELECT active + FROM qiita.software_command + WHERE command_id = %s""" + else: + sql = """SELECT EXISTS ( + SELECT active FROM qiita.software_command + WHERE name IN ( + SELECT name FROM qiita.software_command + WHERE command_id = %s) AND active = true)""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + def activate(self): + """Activates the command""" + sql = """UPDATE qiita.software_command + SET active = %s + WHERE command_id = %s""" + qdb.sql_connection.perform_as_transaction(sql, [True, self.id]) + + @property + def analysis_only(self): + """Returns if the command is an analysis-only command + + Returns + ------- + bool + Whether the command is analysis only or not + """ + with qdb.sql_connection.TRN: + sql = """SELECT is_analysis + FROM qiita.software_command + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def naming_order(self): + """The ordered list of parameters to use to name the output artifacts + + Returns + ------- + list of str + """ + with qdb.sql_connection.TRN: + sql = """SELECT parameter_name + FROM qiita.command_parameter + WHERE command_id = %s AND name_order IS NOT NULL + ORDER BY name_order""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchflatten() + + @property + def merging_scheme(self): + """The values to check when merging the output result + + Returns + ------- + dict of {'parameters': [list of str], + 'outputs': [list of str] + 'ignore_parent_command': bool} + """ + with qdb.sql_connection.TRN: + sql = """SELECT parameter_name + FROM qiita.command_parameter + WHERE command_id = %s AND check_biom_merge = TRUE + ORDER BY parameter_name""" + qdb.sql_connection.TRN.add(sql, [self.id]) + params = qdb.sql_connection.TRN.execute_fetchflatten() + sql = """SELECT name + FROM qiita.command_output + WHERE command_id = %s AND check_biom_merge = TRUE + ORDER BY name""" + qdb.sql_connection.TRN.add(sql, [self.id]) + outputs = qdb.sql_connection.TRN.execute_fetchflatten() + + sql = """SELECT ignore_parent_command + FROM qiita.software_command + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + ipc = qdb.sql_connection.TRN.execute_fetchlast() + + return {'parameters': params, + 'outputs': outputs, + 'ignore_parent_command': ipc} + + @property + def resource_allocation(self): + """The resource allocation defined in the database for this command + + Returns + ------- + str + """ + + with qdb.sql_connection.TRN: + sql = """SELECT allocation FROM + qiita.processing_job_resource_allocation + WHERE name = %s and + job_type = 'RESOURCE_PARAMS_COMMAND'""" + qdb.sql_connection.TRN.add(sql, [self.name]) + + result = qdb.sql_connection.TRN.execute_fetchflatten() + + # if no matches for both type and name were found, query the + # 'default' value for the type + + if not result: + sql = """SELECT allocation FROM + qiita.processing_job_resource_allocation WHERE + name = %s and job_type = 'RESOURCE_PARAMS_COMMAND'""" + qdb.sql_connection.TRN.add(sql, ['default']) + + result = qdb.sql_connection.TRN.execute_fetchflatten() + if not result: + raise ValueError("Could not match '%s' to a resource " + "allocation!" % self.name) + + return result[0] + + @property + def processing_jobs(self): + """All the processing_jobs that used this command + + Returns + ------- + list of qiita_db.processing_job.ProcessingJob + List of jobs that used this command. + """ + + with qdb.sql_connection.TRN: + sql = """SELECT processing_job_id FROM + qiita.processing_job + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + + jids = qdb.sql_connection.TRN.execute_fetchflatten() + + return [qdb.processing_job.ProcessingJob(j) for j in jids] + + +class Software(qdb.base.QiitaObject): + r"""A software package available in the system + + Attributes + ---------- + name + version + description + commands + publications + environment_name + start_script + + Methods + ------- + add_publications + create + + See Also + -------- + qiita_db.software.Command + """ + _table = "software" + + @classmethod + def iter(cls, active=True): + """Iterates over all active software + + Parameters + ---------- + active : bool, optional + If True will only return active software + + Returns + ------- + list of qiita_db.software.Software + The software objects + """ + sql = """SELECT software_id + FROM qiita.software {0} + ORDER BY software_id""".format( + 'WHERE active = True' if active else '') + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add(sql) + for s_id in qdb.sql_connection.TRN.execute_fetchflatten(): + yield cls(s_id) + + @classmethod + def deactivate_all(cls): + """Deactivates all the plugins in the system""" + with qdb.sql_connection.TRN: + sql = "UPDATE qiita.software SET active = False" + qdb.sql_connection.TRN.add(sql) + sql = "UPDATE qiita.software_command SET active = False" + qdb.sql_connection.TRN.add(sql) + qdb.sql_connection.TRN.execute() + + @classmethod + def from_file(cls, fp, update=False): + """Installs/updates a plugin from a plugin configuration file + + Parameters + ---------- + fp : str + Path to the plugin configuration file + update : bool, optional + If true, update the values in the database with the current values + in the config file. Otherwise, use stored values and warn if config + file contents and database contents do not match + + Returns + ------- + qiita_db.software.Software + The software object for the contents of `fp` + + Raises + ------ + qiita_db.exceptions.QiitaDBOperationNotPermittedError + If the plugin type in the DB and in the config file doesn't match + If the (client_id, client_secret) pair in the DB and in the config + file doesn't match + """ + config = ConfigParser() + with open(fp, newline=None) as conf_file: + config.read_file(conf_file) + + name = config.get('main', 'NAME') + version = config.get('main', 'VERSION') + description = config.get('main', 'DESCRIPTION') + env_script = config.get('main', 'ENVIRONMENT_SCRIPT') + start_script = config.get('main', 'START_SCRIPT') + software_type = config.get('main', 'PLUGIN_TYPE') + publications = config.get('main', 'PUBLICATIONS') + publications = loads(publications) if publications else [] + client_id = config.get('oauth2', 'CLIENT_ID') + client_secret = config.get('oauth2', 'CLIENT_SECRET') + + if cls.exists(name, version): + # This plugin already exists, check that all the values are the + # same and return the existing plugin + with qdb.sql_connection.TRN: + sql = """SELECT software_id + FROM qiita.software + WHERE name = %s AND version = %s""" + qdb.sql_connection.TRN.add(sql, [name, version]) + instance = cls(qdb.sql_connection.TRN.execute_fetchlast()) + + warning_values = [] + sql_update = """UPDATE qiita.software + SET {0} = %s + WHERE software_id = %s""" + + values = [description, env_script, start_script] + attrs = ['description', 'environment_script', 'start_script'] + for value, attr in zip(values, attrs): + if value != instance.__getattribute__(attr): + if update: + qdb.sql_connection.TRN.add( + sql_update.format(attr), [value, instance.id]) + else: + warning_values.append(attr) + + # Having a different plugin type should be an error, + # independently if the user is trying to update plugins or not + if software_type != instance.type: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + 'The plugin type of the plugin "%s" version %s does ' + 'not match the one in the system' % (name, version)) + + if publications != instance.publications: + if update: + instance.add_publications(publications) + else: + warning_values.append('publications') + + if (client_id != instance.client_id or + client_secret != instance.client_secret): + if update: + sql = """INSERT INTO qiita.oauth_identifiers + (client_id, client_secret) + SELECT %s, %s + WHERE NOT EXISTS(SELECT * + FROM qiita.oauth_identifiers + WHERE client_id = %s + AND client_secret = %s)""" + qdb.sql_connection.TRN.add( + sql, [client_id, client_secret, + client_id, client_secret]) + sql = """UPDATE qiita.oauth_software + SET client_id = %s + WHERE software_id = %s""" + qdb.sql_connection.TRN.add( + sql, [client_id, instance.id]) + else: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + 'The (client_id, client_secret) pair of the ' + 'plugin "%s" version "%s" does not match the one ' + 'in the system' % (name, version)) + + if warning_values: + warnings.warn( + 'Plugin "%s" version "%s" config file does not match ' + 'with stored information. Check the config file or ' + 'run "qiita plugin update" to update the plugin ' + 'information. Offending values: %s' + % (name, version, ", ".join(sorted(warning_values))), + qdb.exceptions.QiitaDBWarning) + qdb.sql_connection.TRN.execute() + else: + # This is a new plugin, create it + instance = cls.create( + name, version, description, env_script, start_script, + software_type, publications=publications, client_id=client_id, + client_secret=client_secret) + + return instance + + @classmethod + def exists(cls, name, version): + """Returns whether the plugin (name, version) already exists + + Parameters + ---------- + name : str + The name of the plugin + version : str + The version of the plugin + """ + with qdb.sql_connection.TRN: + sql = """SELECT EXISTS( + SELECT * FROM qiita.software + WHERE name = %s AND version = %s)""" + qdb.sql_connection.TRN.add(sql, [name, version]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @classmethod + def create(cls, name, version, description, environment_script, + start_script, software_type, publications=None, + client_id=None, client_secret=None): + r"""Creates a new software in the system + + Parameters + ---------- + name : str + The name of the software + version : str + The version of the software + description : str + The description of the software + environment_script : str + The script used to start the environment in which the plugin runs + start_script : str + The script used to start the plugin + software_type : str + The type of the software + publications : list of (str, str), optional + A list with the (DOI, pubmed_id) of the publications attached to + the software + client_id : str, optional + The client_id of the software. Default: randomly generated + client_secret : str, optional + The client_secret of the software. Default: randomly generated + + Raises + ------ + qiita_db.exceptions.QiitaDBError + If one of client_id or client_secret is provided but not both + """ + with qdb.sql_connection.TRN: + sql = """INSERT INTO qiita.software + (name, version, description, environment_script, + start_script, software_type_id) + VALUES (%s, %s, %s, %s, %s, %s) + RETURNING software_id""" + type_id = qdb.util.convert_to_id(software_type, "software_type") + sql_params = [name, version, description, environment_script, + start_script, type_id] + qdb.sql_connection.TRN.add(sql, sql_params) + s_id = qdb.sql_connection.TRN.execute_fetchlast() + + instance = cls(s_id) + + if publications: + instance.add_publications(publications) + + id_is_none = client_id is None + secret_is_none = client_secret is None + + if id_is_none and secret_is_none: + # Both are none, generate new ones + client_id = qdb.util.create_rand_string(50, punct=False) + client_secret = qdb.util.create_rand_string(255, punct=False) + elif id_is_none ^ secret_is_none: + # One has been provided but not the other, raise an error + raise qdb.exceptions.QiitaDBError( + 'Plugin "%s" version "%s" cannot be created, please ' + 'provide both client_id and client_secret or none of them' + % (name, version)) + + # At this point both client_id and client_secret are defined + sql = """INSERT INTO qiita.oauth_identifiers + (client_id, client_secret) + SELECT %s, %s + WHERE NOT EXISTS(SELECT * + FROM qiita.oauth_identifiers + WHERE client_id = %s + AND client_secret = %s)""" + qdb.sql_connection.TRN.add( + sql, [client_id, client_secret, client_id, client_secret]) + sql = """INSERT INTO qiita.oauth_software (software_id, client_id) + VALUES (%s, %s)""" + qdb.sql_connection.TRN.add(sql, [s_id, client_id]) + + return instance + + @classmethod + def from_name_and_version(cls, name, version): + """Returns the software object with the given name and version + + Parameters + ---------- + name: str + The software name + version : str + The software version + + Returns + ------- + qiita_db.software.Software + The software with the given name and version + + Raises + ------ + qiita_db.exceptions.QiitaDBUnknownIDError + If no software with the given name and version exists + """ + with qdb.sql_connection.TRN: + sql = """SELECT software_id + FROM qiita.software + WHERE name = %s AND version = %s""" + qdb.sql_connection.TRN.add(sql, [name, version]) + res = qdb.sql_connection.TRN.execute_fetchindex() + if not res: + raise qdb.exceptions.QiitaDBUnknownIDError( + "%s %s" % (name, version), cls._table) + return cls(res[0][0]) + + @property + def name(self): + """The name of the software + + Returns + ------- + str + The name of the software + """ + with qdb.sql_connection.TRN: + sql = "SELECT name FROM qiita.software WHERE software_id = %s" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def version(self): + """The version of the software + + Returns + ------- + str + The version of the software + """ + with qdb.sql_connection.TRN: + sql = "SELECT version FROM qiita.software WHERE software_id = %s" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def description(self): + """The description of the software + + Returns + ------- + str + The software description + """ + with qdb.sql_connection.TRN: + sql = """SELECT description + FROM qiita.software + WHERE software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def commands(self): + """The list of commands attached to this software + + Returns + ------- + list of qiita_db.software.Command + The commands attached to this software package + """ + with qdb.sql_connection.TRN: + sql = """SELECT command_id + FROM qiita.software_command + WHERE software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return [Command(cid) + for cid in qdb.sql_connection.TRN.execute_fetchflatten()] + + def get_command(self, cmd_name): + """Returns the command with the given name in the software + + Parameters + ---------- + cmd_name: str + The command with the given name + + Returns + ------- + qiita_db.software.Command + The command with the given name in this software + """ + with qdb.sql_connection.TRN: + sql = """SELECT command_id + FROM qiita.software_command + WHERE software_id =%s AND name=%s""" + qdb.sql_connection.TRN.add(sql, [self.id, cmd_name]) + res = qdb.sql_connection.TRN.execute_fetchindex() + if not res: + raise qdb.exceptions.QiitaDBUnknownIDError( + cmd_name, "software_command") + return Command(res[0][0]) + + @property + def publications(self): + """The publications attached to the software + + Returns + ------- + list of (str, str) + The list of DOI and pubmed_id attached to the publication + """ + with qdb.sql_connection.TRN: + sql = """SELECT p.doi, p.pubmed_id + FROM qiita.publication p + JOIN qiita.software_publication sp + ON p.doi = sp.publication_doi + WHERE sp.software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchindex() + + def add_publications(self, publications): + """Add publications to the software + + Parameters + ---------- + publications : list of 2-tuples of str + A list with the (DOI, pubmed_id) of the publications to be attached + to the software + + Notes + ----- + For more information about pubmed id, visit + https://www.nlm.nih.gov/bsd/disted/pubmedtutorial/020_830.html + """ + with qdb.sql_connection.TRN: + sql = """INSERT INTO qiita.publication (doi, pubmed_id) + SELECT %s, %s + WHERE NOT EXISTS(SELECT * + FROM qiita.publication + WHERE doi = %s)""" + args = [[doi, pid, doi] for doi, pid in publications] + qdb.sql_connection.TRN.add(sql, args, many=True) + + sql = """INSERT INTO qiita.software_publication + (software_id, publication_doi) + SELECT %s, %s + WHERE NOT EXISTS(SELECT * + FROM qiita.software_publication + WHERE software_id = %s AND + publication_doi = %s)""" + sql_params = [[self.id, doi, self.id, doi] + for doi, _ in publications] + qdb.sql_connection.TRN.add(sql, sql_params, many=True) + qdb.sql_connection.TRN.execute() + + @property + def environment_script(self): + """The script used to start the plugin environment + + Returns + ------- + str + The script used to start the environment + """ + with qdb.sql_connection.TRN: + sql = """SELECT environment_script + FROM qiita.software + WHERE software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def start_script(self): + """The script used to start the plugin + + Returns + ------- + str + The plugin's start script + """ + with qdb.sql_connection.TRN: + sql = """SELECT start_script + FROM qiita.software + WHERE software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def type(self): + """Returns the type of the software + + Returns + ------- + str + The type of the software + """ + with qdb.sql_connection.TRN: + sql = """SELECT software_type + FROM qiita.software_type + JOIN qiita.software USING (software_type_id) + WHERE software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def deprecated(self): + """Returns if the software is deprecated or not + + Returns + ------- + bool + Whether the software is deprecated or not + """ + with qdb.sql_connection.TRN: + sql = """SELECT deprecated + FROM qiita.software + WHERE software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @deprecated.setter + def deprecated(self, deprecate): + """Changes deprecated of the software + + Parameters + ---------- + deprecate : bool + New software deprecate value + """ + sql = """UPDATE qiita.software SET deprecated = %s + WHERE software_id = %s""" + qdb.sql_connection.perform_as_transaction(sql, [deprecate, self._id]) + + @property + def active(self): + """Returns if the software is active or not + + Returns + ------- + bool + Whether the software is active or not + """ + with qdb.sql_connection.TRN: + sql = "SELECT active FROM qiita.software WHERE software_id = %s" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + def activate(self): + """Activates the plugin""" + sql = """UPDATE qiita.software + SET active = %s + WHERE software_id = %s""" + qdb.sql_connection.perform_as_transaction(sql, [True, self.id]) + + @property + def client_id(self): + """Returns the client id of the plugin + + Returns + ------- + str + The client id of the software + """ + with qdb.sql_connection.TRN: + sql = """SELECT client_id + FROM qiita.oauth_software + WHERE software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def client_secret(self): + """Returns the client secret of the plugin + + Returns + ------- + str + The client secrect of the plugin + """ + with qdb.sql_connection.TRN: + sql = """SELECT client_secret + FROM qiita.oauth_software + JOIN qiita.oauth_identifiers USING (client_id) + WHERE software_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + def register_commands(self): + """Registers the software commands""" + url = "%s%s" % (qiita_config.base_url, qiita_config.portal_dir) + cmd = '%s; %s "%s" "register" "ignored"' % ( + self.environment_script, self.start_script, url) + + # it can be assumed that any command beginning with 'source' + # is calling 'source', an internal command of 'bash' and hence + # should be executed from bash, instead of sh. + # TODO: confirm that exit_code propagates from bash to sh to + # rv. + if cmd.startswith('source'): + cmd = "bash -c '%s'" % cmd + + p_out, p_err, rv = qdb.processing_job._system_call(cmd) + + if rv != 0: + s = "cmd: %s\nexit status: %d\n" % (cmd, rv) + s += "stdout: %s\nstderr: %s\n" % (p_out, p_err) + + raise ValueError(s) + + +class DefaultParameters(qdb.base.QiitaObject): + """Models a default set of parameters of a command + + Attributes + ---------- + name + values + + Methods + ------- + exists + create + iter + to_str + to_file + + See Also + -------- + qiita_db.software.Command + """ + _table = 'default_parameter_set' + + @classmethod + def exists(cls, command, **kwargs): + r"""Check if a parameter set already exists + + Parameters + ---------- + command : qiita_db.software.Command + The command to which the parameter set belongs to + kwargs : dict of {str: str} + The parameters and their values + + Returns + ------- + bool + Whether if the parameter set exists in the given command + + Raises + ------ + qiita_db.exceptions.QiitaDBError + - If there are missing parameters for the given command + - If `kwargs` contains parameters not originally defined in the + command + """ + with qdb.sql_connection.TRN: + command_params = set(command.optional_parameters) + user_params = set(kwargs) + + missing_in_user = command_params - user_params + extra_in_user = user_params - command_params + + if missing_in_user or extra_in_user: + raise qdb.exceptions.QiitaDBError( + "The given set of parameters do not match the ones for " + "the command.\nMissing parameters: %s\n" + "Extra parameters: %s\n" + % (', '.join(missing_in_user), ', '.join(extra_in_user))) + + sql = """SELECT parameter_set + FROM qiita.default_parameter_set + WHERE command_id = %s""" + qdb.sql_connection.TRN.add(sql, [command.id]) + for p_set in qdb.sql_connection.TRN.execute_fetchflatten(): + if p_set == kwargs: + return True + + return False + + @classmethod + def create(cls, param_set_name, command, **kwargs): + r"""Create a new parameter set for the given command + + Parameters + ---------- + param_set_name: str + The name of the new parameter set + command : qiita_db.software.Command + The command to add the new parameter set + kwargs : dict + The parameters and their values + + Returns + ------- + qiita_db.software.Parameters + The new parameter set instance + + Raises + ------ + qiita_db.exceptions.QiitaDBError + - If there are missing parameters for the given command + - If there are extra parameters in `kwargs` than for the given + command + qdb.exceptions.QiitaDBDuplicateError + - If the parameter set already exists + """ + with qdb.sql_connection.TRN: + # setting to default values all parameters not in the user_params + cmd_params = command.optional_parameters + missing_in_user = {k: cmd_params[k][1] + for k in (set(cmd_params) - set(kwargs))} + if missing_in_user: + kwargs.update(missing_in_user) + + # If the columns in kwargs and command do not match, cls.exists + # will raise the error for us + if cls.exists(command, **kwargs): + raise qdb.exceptions.QiitaDBDuplicateError( + cls._table, "Values: %s" % kwargs) + + sql = """INSERT INTO qiita.default_parameter_set + (command_id, parameter_set_name, parameter_set) + VALUES (%s, %s, %s) + RETURNING default_parameter_set_id""" + sql_args = [command.id, param_set_name, dumps(kwargs)] + qdb.sql_connection.TRN.add(sql, sql_args) + + return cls(qdb.sql_connection.TRN.execute_fetchlast()) + + @property + def name(self): + """The name of the parameter set + + Returns + ------- + str + The name of the parameter set + """ + with qdb.sql_connection.TRN: + sql = """SELECT parameter_set_name + FROM qiita.default_parameter_set + WHERE default_parameter_set_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def values(self): + """The values of the parameter set + + Returns + ------- + dict of {str: object} + Dictionary with the parameters values keyed by parameter name + """ + with qdb.sql_connection.TRN: + sql = """SELECT parameter_set + FROM qiita.default_parameter_set + WHERE default_parameter_set_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def command(self): + """The command that this parameter set belongs to + + Returns + ------- + qiita_db.software.Command + The command that this parameter set belongs to + """ + with qdb.sql_connection.TRN: + sql = """SELECT command_id + FROM qiita.default_parameter_set + WHERE default_parameter_set_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return Command(qdb.sql_connection.TRN.execute_fetchlast()) + + +class Parameters(object): + """Represents an instance of parameters used to process an artifact + + Raises + ------ + qiita_db.exceptions.QiitaDBOperationNotPermittedError + If trying to instantiate this class directly. In order to instantiate + this class, the classmethods `load` or `from_default_params` should + be used. + """ + + def __eq__(self, other): + """Equality based on the parameter values and the command""" + if type(self) is not type(other): + return False + if self.command != other.command: + return False + if self.values != other.values: + return False + return True + + @classmethod + def load(cls, command, json_str=None, values_dict=None): + """Load the parameters set form a json str or from a dict of values + + Parameters + ---------- + command : qiita_db.software.Command + The command to which the parameter set belongs to + json_str : str, optional + The json string encoding the parameters + values_dict : dict of {str: object}, optional + The dictionary with the parameter values + + Returns + ------- + qiita_db.software.Parameters + The loaded parameter set + + Raises + ------ + qiita_db.exceptions.QiitaDBError + - If `json_str` and `values` are both provided + - If neither `json_str` or `values` are provided + - If `json_str` or `values` do not encode a parameter set of + the provided command. + + Notes + ----- + The parameters `json_str` and `values_dict` are mutually exclusive, + only one of them should be provided at a time. However, one of them + should always be provided. + """ + if json_str is None and values_dict is None: + raise qdb.exceptions.QiitaDBError( + "Either `json_str` or `values_dict` should be provided.") + elif json_str is not None and values_dict is not None: + raise qdb.exceptions.QiitaDBError( + "Either `json_str` or `values_dict` should be provided, " + "but not both") + elif json_str is not None: + parameters = loads(json_str) + error_msg = ("The provided JSON string doesn't encode a " + "parameter set for command %s" % command.id) + else: + if not isinstance(values_dict, dict): + raise qdb.exceptions.QiitaDBError( + "The provided value_dict is %s (i.e. not None) but also " + "not a dictionary for command %s" % ( + values_dict, command.id)) + parameters = deepcopy(values_dict) + error_msg = ("The provided values dictionary doesn't encode a " + "parameter set for command %s" % command.id) + + # setting to default values all parameters not in the user_params + cmd_params = command.optional_parameters + missing_in_user = {k: cmd_params[k][1] + for k in (set(cmd_params) - set(parameters))} + if missing_in_user: + parameters.update(missing_in_user) + + with qdb.sql_connection.TRN: + cmd_reqd_params = command.required_parameters + cmd_opt_params = command.optional_parameters + + values = {} + for key in cmd_reqd_params: + try: + values[key] = parameters.pop(key) + except KeyError: + raise qdb.exceptions.QiitaDBError( + "%s. Missing required parameter: %s" + % (error_msg, key)) + + for key in cmd_opt_params: + try: + values[key] = parameters.pop(key) + except KeyError: + raise qdb.exceptions.QiitaDBError( + "%s. Missing optional parameter: %s" + % (error_msg, key)) + + if parameters: + raise qdb.exceptions.QiitaDBError( + "%s. Extra parameters: %s" + % (error_msg, ', '.join(parameters.keys()))) + + return cls(values, command) + + @classmethod + def from_default_params(cls, dflt_params, req_params, opt_params=None): + """Creates the parameter set from a `dflt_params` set + + Parameters + ---------- + dflt_params : qiita_db.software.DefaultParameters + The DefaultParameters object in which this instance is based on + req_params : dict of {str: object} + The required parameters values, keyed by parameter name + opt_params : dict of {str: object}, optional + The optional parameters to change from the default set, keyed by + parameter name. Default: None, use the values in `dflt_params` + + Raises + ------ + QiitaDBError + - If there are missing requried parameters + - If there is an unknown required ot optional parameter + """ + with qdb.sql_connection.TRN: + command = dflt_params.command + cmd_req_params = command.required_parameters + cmd_opt_params = command.optional_parameters + + missing_reqd = set(cmd_req_params) - set(req_params) + extra_reqd = set(req_params) - set(cmd_req_params) + if missing_reqd or extra_reqd: + raise qdb.exceptions.QiitaDBError( + "Provided required parameters not expected.\n" + "Missing required parameters: %s\n" + "Extra required parameters: %s\n" + % (', '.join(missing_reqd), ', '.join(extra_reqd))) + + if opt_params: + extra_opts = set(opt_params) - set(cmd_opt_params) + if extra_opts: + raise qdb.exceptions.QiitaDBError( + "Extra optional parameters provded: %s" + % ', '.join(extra_opts)) + + values = dflt_params.values + values.update(req_params) + + if opt_params: + values.update(opt_params) + + return cls(values, command) + + def __init__(self, values, command): + # Time for some python magic! The __init__ function should not be used + # outside of this module, users should always be using one of the above + # classmethods to instantiate the object. Lets test that it is the case + # First, we are going to get the current frame (i.e. this __init__) + # function and the caller to the __init__ + current_frame = inspect.currentframe() + caller_frame = current_frame.f_back + # The file names where the function is defined is stored in the + # f_code.co_filename attribute, and in this case it has to be the same + # for both of them. Also, we are restricing that the name of the caller + # should be either `load` or `from_default_params`, which are the two + # classmethods defined above + current_file = current_frame.f_code.co_filename + caller_file = caller_frame.f_code.co_filename + caller_name = caller_frame.f_code.co_name + if current_file != caller_file or \ + caller_name not in ['load', 'from_default_params']: + raise qdb.exceptions.QiitaDBOperationNotPermittedError( + "qiita_db.software.Parameters can't be instantiated directly. " + "Please use one of the classmethods: `load` or " + "`from_default_params`") + + self._values = values + self._command = command + + @property + def command(self): + """The command to which this parameter set belongs to + + Returns + ------- + qiita_db.software.Command + The command to which this parameter set belongs to + """ + return self._command + + @property + def values(self): + """The values of the parameters + + Returns + ------- + dict of {str: object} + The parameter values keyed by parameter name + """ + return self._values + + def dump(self): + """Return the values in the parameter as JSON + + Returns + ------- + str + The parameter values as a JSON string + """ + return dumps(self._values, sort_keys=True) + + +class DefaultWorkflowNode(qdb.base.QiitaObject): + r"""Represents a node in a default software workflow + + Attributes + ---------- + command + parameters + """ + _table = "default_workflow_node" + + @property + def default_parameter(self): + """The default parameter set to use in this node + + Returns + ------- + qiita_db.software.DefaultParameters + """ + with qdb.sql_connection.TRN: + sql = """SELECT default_parameter_set_id + FROM qiita.default_workflow_node + WHERE default_workflow_node_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + params_id = qdb.sql_connection.TRN.execute_fetchlast() + return qdb.software.DefaultParameters(params_id) + + +class DefaultWorkflowEdge(qdb.base.QiitaObject): + r"""Represents an edge in a default software workflow + + Attributes + ---------- + connections + """ + _table = "default_workflow_edge" + + @property + def connections(self): + """Retrieve how the commands are connected using this edge + + Returns + ------- + list of [str, str] + The list of pairs of output parameter name and input parameter name + used to connect the output of the source command to the input of + the destination command. + """ + with qdb.sql_connection.TRN: + sql = """SELECT name, parameter_name, artifact_type + FROM qiita.default_workflow_edge_connections c + JOIN qiita.command_output o + ON c.parent_output_id = o.command_output_id + JOIN qiita.command_parameter p + ON c.child_input_id = p.command_parameter_id + LEFT JOIN qiita.artifact_type USING (artifact_type_id) + WHERE default_workflow_edge_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchindex() + + +class DefaultWorkflow(qdb.base.QiitaObject): + r"""Represents a software's default workflow + + A default workflow is defined by a Directed Acyclic Graph (DAG) in which + the nodes represent the commands to be executed with the default parameter + set to use and the edges represent the command precedence, including + which outputs of the source command are provided as input to the + destination command. + """ + _table = "default_workflow" + + @classmethod + def iter(cls, active=True): + """Iterates over all active DefaultWorkflow + + Parameters + ---------- + active : bool, optional + If True will only return active software + + Returns + ------- + list of qiita_db.software.DefaultWorkflow + The DefaultWorkflow objects + """ + sql = """SELECT default_workflow_id + FROM qiita.default_workflow {0} + ORDER BY default_workflow_id""".format( + 'WHERE active = True' if active else '') + with qdb.sql_connection.TRN: + qdb.sql_connection.TRN.add(sql) + for s_id in qdb.sql_connection.TRN.execute_fetchflatten(): + yield cls(s_id) + + @property + def active(self): + """Retrieves active status of the default workflow + + Returns + ------- + active : bool + active value + """ + with qdb.sql_connection.TRN: + sql = """SELECT active + FROM qiita.default_workflow + WHERE default_workflow_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @active.setter + def active(self, active): + """Changes active status of the default workflow + + Parameters + ---------- + active : bool + New active value + """ + sql = """UPDATE qiita.default_workflow SET active = %s + WHERE default_workflow_id = %s""" + qdb.sql_connection.perform_as_transaction(sql, [active, self._id]) + + @property + def name(self): + with qdb.sql_connection.TRN: + sql = """SELECT name + FROM qiita.default_workflow + WHERE default_workflow_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @property + def description(self): + """Retrieves the description of the default workflow + + Returns + ------- + str + description value + """ + with qdb.sql_connection.TRN: + sql = """SELECT description + FROM qiita.default_workflow + WHERE default_workflow_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchlast() + + @description.setter + def description(self, description): + """Changes the description of the default workflow + + Parameters + ---------- + description : str + New description value + """ + sql = """UPDATE qiita.default_workflow SET description = %s + WHERE default_workflow_id = %s""" + qdb.sql_connection.perform_as_transaction(sql, [description, self._id]) + + @property + def data_type(self): + """Retrieves all the data_types accepted by the default workflow + + Returns + ---------- + list of str + The data types + """ + with qdb.sql_connection.TRN: + sql = """SELECT data_type + FROM qiita.default_workflow_data_type + LEFT JOIN qiita.data_type USING (data_type_id) + WHERE default_workflow_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchflatten() + + @property + def artifact_type(self): + """Retrieves artifact_type that the workflow can be applied to + + Returns + ---------- + str + The name of the artifact type this workflow can be applied to + """ + with qdb.sql_connection.TRN: + sql = """SELECT artifact_type + FROM qiita.artifact_type + LEFT JOIN qiita.default_workflow USING (artifact_type_id) + WHERE default_workflow_id = %s""" + qdb.sql_connection.TRN.add(sql, [self.id]) + return qdb.sql_connection.TRN.execute_fetchflatten()[0] + + @property + def graph(self): + """Returns the graph that represents the workflow + + Returns + ------- + networkx.DiGraph + The graph representing the default workflow. + """ + g = nx.DiGraph() + with qdb.sql_connection.TRN: + # Retrieve all graph workflow nodes + sql = """SELECT default_workflow_node_id + FROM qiita.default_workflow_node + WHERE default_workflow_id = %s + ORDER BY default_workflow_node_id""" + qdb.sql_connection.TRN.add(sql, [self.id]) + db_nodes = qdb.sql_connection.TRN.execute_fetchflatten() + + nodes = {n_id: DefaultWorkflowNode(n_id) for n_id in db_nodes} + + # Retrieve all graph edges + sql = """SELECT DISTINCT default_workflow_edge_id, parent_id, + child_id + FROM qiita.default_workflow_edge e + JOIN qiita.default_workflow_node n + ON e.parent_id = n.default_workflow_node_id + OR e.child_id = n.default_workflow_node_id + WHERE default_workflow_id = %s + ORDER BY default_workflow_edge_id""" + qdb.sql_connection.TRN.add(sql, [self.id]) + db_edges = qdb.sql_connection.TRN.execute_fetchindex() + + # let's track what nodes are actually being used so if they do not + # have an edge we still return them as part of the graph + used_nodes = nodes.copy() + for edge_id, p_id, c_id in db_edges: + e = DefaultWorkflowEdge(edge_id) + g.add_edge(nodes[p_id], nodes[c_id], connections=e) + if p_id in used_nodes: + del used_nodes[p_id] + if c_id in used_nodes: + del used_nodes[c_id] + # adding the missing nodes + for ms in used_nodes: + g.add_node(nodes[ms]) + + 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])