Source code for galaxy.managers.executables

"""Utilities for loading tools and workflows from paths for admin user requests."""

from typing import (
    Any,
    Dict,
    Optional,
)

import yaml

from galaxy import exceptions
from galaxy.util import in_directory


[docs]def artifact_class(trans, as_dict: Dict[str, Any], allow_in_directory: Optional[str] = None):
    object_id = as_dict.get("object_id", None)
    if as_dict.get("src", None) == "from_path":
        workflow_path = as_dict.get("path")
        allow = not trans or trans.user_is_admin
        allow = allow or (allow_in_directory and in_directory(workflow_path, allow_in_directory))

        if not allow:
            raise exceptions.AdminRequiredException()

        if not isinstance(workflow_path, str):
            raise exceptions.RequestParameterInvalidException()

        with open(workflow_path) as f:
            as_dict = yaml.safe_load(f)

    artifact_class = as_dict.get("class", None)
    if artifact_class is None and "$graph" in as_dict:
        object_id = object_id or "main"
        graph = as_dict["$graph"]
        target_object = None
        if isinstance(graph, dict):
            target_object = graph.get(object_id)
        else:
            for item in graph:
                found_id = item.get("id")
                if found_id == object_id or found_id == f"#{object_id}":
                    target_object = item

        if target_object and target_object.get("class"):
            artifact_class = target_object["class"]

    return artifact_class, as_dict, object_id


__all__ = ("artifact_class",)