Source code for planemo.autoupdate

"""Autoupdate older conda dependencies in the requirements section."""

import collections
import itertools
import re
import xml.etree.ElementTree as ET
from typing import (
from xml.etree.ElementTree import ElementTree

import requests
import yaml
from bioblend import toolshed
from bioblend.toolshed import ToolShedInstance
from galaxy.tool_util.deps import conda_util
from galaxy.tool_util.version import parse_version

import planemo.conda
from import (
from planemo.workflow_lint import (

    from planemo.cli import PlanemoCliContext
    from planemo.galaxy.config import LocalGalaxyConfig
    from planemo.runnable import Runnable

[docs] def find_macros(xml_tree: ElementTree) -> List[Any]: """ Get macros from the XML tree """ macros = [] for macro_import in xml_tree.iter("import"): macros.append(macro_import.text) return macros
[docs] def get_requirements(xml_tree: ElementTree) -> Tuple[Dict[str, Dict[str, Optional[str]]], Optional[str]]: """ Get requirements from the XML tree """ requirements = {} main_req = None for requirement in xml_tree.iter("requirement"): if requirement.attrib.get("version") == "@TOOL_VERSION@": main_req = requirement.text else: assert requirement.text requirements[requirement.text] = { "tag": ET.tostring(requirement, encoding="unicode").strip(), "text": requirement.attrib.get("version"), } return requirements, main_req
[docs] def get_tokens(xml_tree: ElementTree) -> Dict[str, Dict[str, Optional[str]]]: """ Get tokens from the XML tree """ tokens = {} for token in xml_tree.iter("token"): tokens[token.attrib["name"]] = {"tag": ET.tostring(token, encoding="unicode").strip(), "text": token.text} return tokens
[docs] def check_conda(package_name: str, ctx: "PlanemoCliContext", **kwds) -> str: """ Get the most up-to-date conda version for a package. """ conda_context = planemo.conda.build_conda_context(ctx, **kwds) if not conda_context.is_conda_installed(): # check directly via Anaconda API r = requests.get("", params={"name": package_name}) search_results = itertools.chain.from_iterable( n["versions"] for n in r.json() if n["name"] == package_name and n["owner"] in kwds["conda_ensure_channels"] ) return sorted(search_results, key=parse_version, reverse=True)[0] target = conda_util.CondaTarget(package_name) best_search_results = conda_util.best_search_result(target, conda_context=conda_context) if best_search_results[0] is None: raise Exception(f"No conda package found for {package_name}") return best_search_results[0]["version"]
[docs] def update_xml( tool_path: str, xml_tree: ElementTree, tags_to_update: List[Dict[str, str]], wrapper_version_token: Optional[Union[int, str]], is_macro: bool = False, ) -> None: """ Write modified XML to tool_path """ def update_token(xml_text, tag, token_value): new_tag = f">{token_value}<".join(re.split(">.*<", tag)) return re.sub(tag, new_tag, xml_text) def update_requirement(xml_text, tag, requirement_value): new_tag = f'version="{requirement_value}"'.join(re.split('version=".*"', tag)) return re.sub(tag, new_tag, xml_text) with open(tool_path, "r+", newline="") as f: xml_text = for tag_to_update in tags_to_update: if tag_to_update["type"] == "token": xml_text = update_token(xml_text, tag_to_update["tag"], tag_to_update["value"]) if tag_to_update["type"] == "requirement": xml_text = update_requirement(xml_text, tag_to_update["tag"], tag_to_update["value"]) if wrapper_version_token == 0 and not is_macro: # i.e. @VERSION_SUFFIX@ not specified so update the version directly in the tool tag tool_tag = re.sub( 'version="@TOOL_VERSION@.*?"', 'version="@TOOL_VERSION@+galaxy0"', re.findall('<tool .*version="@TOOL_VERSION@.*">', xml_text)[0], ) xml_text = re.sub('<tool .*version="@TOOL_VERSION@.*">', tool_tag, xml_text) f.truncate() f.write(xml_text)
[docs] def create_requirement_dict( xml_files: Dict[str, ElementTree], skip_reqs: List[str] ) -> Tuple[Dict[str, Dict[str, Dict[str, Optional[str]]]], Optional[Tuple[str, str]]]: """ Create dict with requirements and find main requirement """ requirements = {} main_req = None for k, v in xml_files.items(): file_reqs, file_main_req = get_requirements(v) requirements[k] = {k: v for k, v in file_reqs.items() if k not in skip_reqs} if file_main_req: if main_req: error("Multiple requirements use the token @TOOL_VERSION@!") main_req = (file_main_req, k) if not main_req: error("No requirement uses the token @TOOL_VERSION@!") return requirements, main_req
[docs] def create_token_dict( ctx: "PlanemoCliContext", xml_files: Dict[str, ElementTree], main_req: Tuple[str, str], **kwds ) -> Tuple[ Dict[str, Dict[str, Dict[str, Optional[str]]]], DefaultDict[str, List[Dict[str, str]]], Optional[str], Optional[str] ]: """ Create dict with relevant tokens and check conda requirements for main """ tokens: Dict[str, Dict[str, Dict[str, Optional[str]]]] = {} current_main_req, updated_main_req = None, None xml_to_update = collections.defaultdict(list) for k, v in xml_files.items(): tokens[k] = get_tokens(v) # check if it is @TOOL_VERSION@ and if so do check_conda if "@TOOL_VERSION@" in tokens[k]: current_main_req = tokens[k]["@TOOL_VERSION@"]["text"] updated_main_req = check_conda(main_req[0], ctx, **kwds) if current_main_req: tag = tokens[k]["@TOOL_VERSION@"]["tag"] assert tag is not None xml_to_update[k].append({"type": "token", "tag": tag, "value": updated_main_req}) return tokens, xml_to_update, current_main_req, updated_main_req
[docs] def perform_required_update( ctx: "PlanemoCliContext", xml_files: Dict[str, ElementTree], tool_path: str, requirements: Dict[str, Dict[str, Dict[str, Optional[str]]]], tokens: Dict[str, Dict[str, Dict[str, Optional[str]]]], xml_to_update: DefaultDict[str, List[Dict[str, str]]], wrapper_version_token: Optional[Union[int, str]], **kwds, ) -> Set[str]: """ Carry out the update, if requirements are out-of-date """ # check all requirements for k, v in requirements.items(): for req in v: req_check = check_conda(req, ctx, **kwds) # print(req_check, v[req]['text']) if req_check != v[req]["text"]: tag = v[req]["tag"] assert tag is not None xml_to_update[k].append({"type": "requirement", "tag": tag, "value": req_check}) # check all tokens, if wrapper_version_token exists if wrapper_version_token: for k, v in tokens.items(): if isinstance(wrapper_version_token, str) and wrapper_version_token in v: tag = v[wrapper_version_token]["tag"] assert tag is not None xml_to_update[k].append({"type": "token", "tag": tag, "value": "0"}) # finally, update each file separately for k, et in xml_files.items(): update_xml(k, et, xml_to_update[k], wrapper_version_token, is_macro=(k != tool_path)) info(f"Tool {tool_path} successfully updated.") return set(xml_files)
[docs] def autoupdate_tool(ctx: "PlanemoCliContext", tool_path: str, modified_files: Set[Any], **kwds) -> Optional[Set[str]]: """ Autoupdate an XML file """ modified_files = modified_files or set() # create a dict of all files that need editing - wrapper plus macros xml_files = {tool_path: ET.parse(tool_path)} # get name of token which defines the wrapper version; if just an integer, None versions = xml_files[tool_path].getroot().attrib.get("version") if versions: versions = versions.split("+galaxy") if versions[0] != "@TOOL_VERSION@": error("Tool version does not contain @TOOL_VERSION@ as required by autoupdate.") return None elif len(versions) == 1: wrapper_version_token = None else: if versions[1][0] == versions[1][-1] == "@": wrapper_version_token = versions[1] else: wrapper_version_token = 0 # assume an int, reset to 0 else: wrapper_version_token = None # add macros to xml_files for macro in find_macros(xml_files[tool_path]): macro_path = "/".join(tool_path.split("/")[:-1] + [macro]) xml_files[macro_path] = ET.parse(macro_path) requirements, main_req = create_requirement_dict(xml_files, kwds.get("skip_requirements", "").split(",")) if main_req is None: return None tokens, xml_to_update, current_main_req, updated_main_req = create_token_dict(ctx, xml_files, main_req, **kwds) if current_main_req == updated_main_req and not (modified_files & set(xml_files)): info(f"No updates required or made to {tool_path}.") return None # end here if no update needed if kwds.get("dry_run"): error( f"Update required to {tool_path}! Tool main requirement has version {current_main_req}, newest conda version is {updated_main_req}" ) return None else: info(f"Updating {tool_path.split('/')[-1]} from version {current_main_req} to {updated_main_req}") return perform_required_update( ctx, xml_files, tool_path, requirements, tokens, xml_to_update, wrapper_version_token, **kwds )
def _update_wf(config: "LocalGalaxyConfig", workflow_id: str, instance: bool = False) -> None: """ Recursively update a workflow, including subworkflows """ wf = config.user_gi.make_get_request( f"{config.user_gi.url}/workflows/{workflow_id}", params={"instance": instance} ).json() for step in wf.get("steps", {}).values(): if step["type"] == "subworkflow": # update subworkflows before the main workflow _update_wf(config, step["workflow_id"], instance=True) config.user_gi.workflows.refactor_workflow(wf["id"], actions=[{"action_type": "upgrade_all_steps"}])
[docs] def get_newest_tool_id(tool_ids: List[str]) -> str: return sorted( tool_ids, key=lambda n: parse_version(n.split("/")[-1]), )[-1]
[docs] def outdated_tools( # noqa: C901 ctx: "PlanemoCliContext", wf_dict: Dict[str, Any], ts: ToolShedInstance, tools_to_skip: List[str] ) -> Dict[str, Dict[str, str]]: """ tools_to_skip should be a list of base tool ids. """ def base_tool_id(tool_id: str) -> str: return tool_id.rsplit("/", 1)[0] def check_tool_step(tool_id: str) -> Dict[str, Dict[str, str]]: """ Return a dict with current and newest tool version, in case they don't match """ warning_msg, repos = find_repos_from_tool_id(tool_id, ts) if warning_msg != "": ctx.log(warning_msg) if len(repos) == 0: return repos base_id = base_tool_id(tool_id) matching_tool_ids = [] for repo in repos.values(): if isinstance(repo, dict): for tool in repo.get("tools") or []: if tool["guid"].startswith(base_id + "/"): matching_tool_ids.append(tool["guid"]) # there can only be one matching tool id in a repo break updated_tool_id = get_newest_tool_id(matching_tool_ids) if tool_id != updated_tool_id: return {base_id: {"current": tool_id, "updated": updated_tool_id}} else: return {} def outdated_tools_rec(wf_dict: Dict[str, Any]) -> None: steps = wf_dict["steps"].values() if isinstance(wf_dict["steps"], dict) else wf_dict["steps"] for step in steps: if step.get("type", "tool") == "tool" and not step.get("run", {}).get("class") == "GalaxyWorkflow": tool_id = step["tool_id"] base_id = base_tool_id(tool_id) if base_id not in checked_tools: outdated_tool_dict.update(check_tool_step(tool_id)) checked_tools.append(base_id) elif step.get("type") == "subworkflow": # GA SWF outdated_tools_rec(step["subworkflow"]) elif step.get("run", {}).get("class") == "GalaxyWorkflow": # gxformat2 SWF outdated_tools_rec(step["run"]) else: continue outdated_tool_dict: Dict[str, Dict[str, str]] = {} # Initialize the list of tools already checked with a copy of tools_to_skip checked_tools = tools_to_skip.copy() outdated_tools_rec(wf_dict) return outdated_tool_dict
[docs] def get_tools_to_update( ctx: "PlanemoCliContext", workflow: "Runnable", tools_to_skip: List[str] ) -> Dict[str, Dict[str, str]]: # before we run the autoupdate, we check the tools against the toolshed to see if there # are any new versions. This saves spinning up Galaxy and installing the tools if there # is nothing to do, and also allows us to collect a list of the tools which need updating with open(workflow.path) as f: wf_dict = yaml.load(f, Loader=yaml.SafeLoader) ts = toolshed.ToolShedInstance(url=MAIN_TOOLSHED_URL) return outdated_tools(ctx, wf_dict, ts, tools_to_skip)
[docs] def autoupdate_wf(ctx: "PlanemoCliContext", config: "LocalGalaxyConfig", wf: "Runnable") -> Dict[str, Any]: workflow_id = config.workflow_id_for_runnable(wf) _update_wf(config, workflow_id) return config.user_gi.workflows.export_workflow_dict(workflow_id)
[docs] def fix_workflow_ga(original_wf: Dict[str, Any], updated_wf: Dict[str, Any]) -> Dict[str, Any]: # the Galaxy refactor action can't do everything right now... some manual fixes here # * bump release number if present # * order steps numerically, leave everything else sorted as in the original workflow # * recurse over subworkflows edited_wf = original_wf.copy() updated_wf_steps = collections.OrderedDict(sorted(updated_wf["steps"].items(), key=lambda item: int(item[0]))) edited_wf["steps"] = updated_wf_steps # check release; bump if it exists if edited_wf.get("release"): release = [int(n) for n in edited_wf["release"].split(".")] release[-1] += 1 edited_wf["release"] = ".".join([str(n) for n in release]) # iterate over the steps for step in edited_wf["steps"]: # recurse over subworkflows if edited_wf["steps"][step].get("type") == "subworkflow": edited_wf["steps"][step]["subworkflow"] = fix_workflow_ga( edited_wf["steps"][step]["subworkflow"], updated_wf["steps"][step]["subworkflow"] ) return edited_wf
[docs] def fix_workflow_gxformat2(original_wf: Dict[str, Any], updated_wf: Dict[str, Any]) -> Dict[str, Any]: # does the same as fix_workflow_ga for gxformat2 edited_wf = original_wf.copy() # check release; bump if it exists if edited_wf.get("release"): release = [int(n) for n in edited_wf["release"].split(".")] release[-1] += 1 edited_wf["release"] = ".".join([str(n) for n in release]) # iterate over the steps for step_index, step in enumerate(edited_wf["steps"]): # recurse over subworkflows if step.get("run", {}).get("class") == "GalaxyWorkflow": # subworkflow step["run"] = fix_workflow_gxformat2( step["run"], updated_wf["steps"][str(step_index + len(original_wf["inputs"]))]["subworkflow"] ) # fix tool_id and content_id to march tool_version elif updated_wf["steps"][str(step_index + len(original_wf["inputs"]))]["type"] == "tool": if ( updated_wf["steps"][str(step_index + len(original_wf["inputs"]))] .get("tool_id", "") .startswith(MAIN_TOOLSHED_URL[8:]) ): step["tool_version"] = updated_wf["steps"][str(step_index + len(original_wf["inputs"]))]["tool_version"] step["tool_id"] = updated_wf["steps"][str(step_index + len(original_wf["inputs"]))]["tool_id"] step["content_id"] = updated_wf["steps"][str(step_index + len(original_wf["inputs"]))]["content_id"] return edited_wf