Source code for galaxy.datatypes.display_applications.parameters

# Contains parameters that are used in Display Applications
import mimetypes
from dataclasses import dataclass
from typing import (
    Callable,
    Optional,
    TYPE_CHECKING,
    Union,
)
from urllib.parse import quote_plus

from galaxy.datatypes.data import Data
from galaxy.model import DatasetInstance
from galaxy.schema.schema import DatasetState
from galaxy.util import string_as_bool
from galaxy.util.template import fill_template

if TYPE_CHECKING:
    from galaxy.datatypes.registry import Registry
DEFAULT_DATASET_NAME = "dataset"


[docs]class DisplayApplicationParameter:
    """Abstract Class for Display Application Parameters"""

    type: Optional[str] = None

[docs]    @classmethod
    def from_elem(cls, elem, link):
        param_type = elem.get("type", None)
        assert param_type, "DisplayApplicationParameter requires a type"
        return parameter_type_to_class[param_type](elem, link)

[docs]    def __init__(self, elem, link):
        self.name = elem.get("name", None)
        assert self.name, "DisplayApplicationParameter requires a name"
        self.link = link
        self.url = elem.get(
            "url", self.name
        )  # name used in url for display purposes defaults to name; e.g. want the form of file.ext, where a '.' is not allowed as python variable name/keyword
        self.mime_type = elem.get("mimetype", None)
        self.guess_mime_type = string_as_bool(elem.get("guess_mimetype", "False"))
        self.viewable = string_as_bool(
            elem.get("viewable", "False")
        )  # only allow these to be viewed via direct url when explicitly set to viewable
        self.strip = string_as_bool(elem.get("strip", "False"))
        self.strip_https = string_as_bool(elem.get("strip_https", "False"))
        self.allow_override = string_as_bool(
            elem.get("allow_override", "False")
        )  # Passing query param app_<name>=<value> to dataset controller allows override if this is true.
        self.allow_cors = string_as_bool(elem.get("allow_cors", "False"))

[docs]    def get_value(self, other_values, dataset_hash, user_hash, trans):
        raise Exception("get_value() is unimplemented for DisplayApplicationDataParameter")

[docs]    def prepare(self, other_values, dataset_hash, user_hash, trans):
        return self.get_value(other_values, dataset_hash, user_hash, trans)

[docs]    def ready(self, other_values):
        return True

[docs]    def is_preparing(self, other_values):
        return False

[docs]    def build_url(self, other_values):
        return fill_template(self.url, context=other_values)


[docs]@dataclass
class DatasetLikeObject:
    get_file_name: Callable
    state: DatasetState
    extension: str
    name: str
    dbkey: Optional[str]
    datatype: Data


[docs]class DisplayApplicationDataParameter(DisplayApplicationParameter):
    """Parameter that returns a file_name containing the requested content"""

    type = "data"

[docs]    def __init__(self, elem, link):
        DisplayApplicationParameter.__init__(self, elem, link)
        self.extensions = elem.get("format", None)
        if self.extensions:
            self.extensions = self.extensions.split(",")
        self.metadata = elem.get("metadata", None)
        self.allow_extra_files_access = string_as_bool(elem.get("allow_extra_files_access", "False"))
        self.dataset = elem.get(
            "dataset", DEFAULT_DATASET_NAME
        )  # 'dataset' is default name assigned to dataset to be displayed
        assert not (
            self.extensions and self.metadata
        ), "A format or a metadata can be defined for a DisplayApplicationParameter, but not both."
        assert not (
            self.allow_extra_files_access and self.metadata
        ), "allow_extra_files_access or metadata can be defined for a DisplayApplicationParameter, but not both."
        self.viewable = string_as_bool(elem.get("viewable", "True"))  # data params should be viewable
        self.force_url_param = string_as_bool(elem.get("force_url_param", "False"))
        self.force_conversion = string_as_bool(elem.get("force_conversion", "False"))

    @property
    def datatypes_registry(self) -> "Registry":
        return self.link.display_application.app.datatypes_registry

    @property
    def formats(self):
        if self.extensions:
            return tuple(
                map(
                    type,
                    map(self.datatypes_registry.get_datatype_by_extension, self.extensions),
                )
            )
        return None

    def _get_dataset_like_object(self, other_values) -> Optional[Union[DatasetLikeObject, DatasetInstance]]:
        data = other_values.get(self.dataset, None)
        assert data, "Base dataset could not be found in values provided to DisplayApplicationDataParameter"
        if isinstance(data, DisplayDataValueWrapper):
            data = data.value
        if data.state != DatasetState.OK:
            return None
        if self.metadata:
            rval = getattr(data.metadata, self.metadata, None)
            if not rval:
                # May have to look at converted datasets
                for converted_dataset_association in data.implicitly_converted_datasets:
                    converted_dataset = converted_dataset_association.dataset
                    if converted_dataset.state != DatasetState.OK:
                        return None
                    rval = getattr(converted_dataset.metadata, self.metadata, None)
                    if rval:
                        data = converted_dataset
                        break
            assert rval, f'Unknown metadata name "{self.metadata}" provided for dataset type "{data.ext}".'
            return DatasetLikeObject(
                get_file_name=rval.get_file_name,
                state=data.state,
                extension="data",
                dbkey=data.dbkey,
                name=data.name,
                datatype=data.datatype,
            )
        elif (
            self.formats and self.extensions and (self.force_conversion or not isinstance(data.datatype, self.formats))
        ):
            for ext in self.extensions:
                rval = data.get_converted_files_by_type(ext)
                if rval:
                    return rval
            direct_match, target_ext, _ = self.datatypes_registry.find_conversion_destination_for_dataset_by_extensions(
                data.extension, self.extensions
            )
            assert direct_match or target_ext is not None, f"No conversion path found for data param: {self.name}"
            return None
        return data

[docs]    def get_value(self, other_values, dataset_hash, user_hash, trans):
        if data := self._get_dataset_like_object(other_values):
            return DisplayDataValueWrapper(data, self, other_values, dataset_hash, user_hash, trans)
        return None

[docs]    def prepare(self, other_values, dataset_hash, user_hash, trans):
        data = self._get_dataset_like_object(other_values)
        if not data and self.formats:
            data = other_values.get(self.dataset, None)
            # start conversion
            # FIXME: Much of this is copied (more than once...); should be some abstract method elsewhere called from here
            # find target ext
            (
                direct_match,
                target_ext,
                converted_dataset,
            ) = self.datatypes_registry.find_conversion_destination_for_dataset_by_extensions(
                data.extension, self.extensions
            )
            if not direct_match:
                if target_ext and not converted_dataset:
                    if isinstance(data, DisplayDataValueWrapper):
                        data = data.value
                    data.datatype.convert_dataset(trans, data, target_ext, return_output=True, visible=False)
                elif converted_dataset and converted_dataset.state == DatasetState.ERROR:
                    raise Exception(f"Dataset conversion failed for data parameter: {self.name}")
        return self.get_value(other_values, dataset_hash, user_hash, trans)

[docs]    def is_preparing(self, other_values):
        value = self._get_dataset_like_object(other_values)
        if value and value.state in (DatasetState.NEW, DatasetState.UPLOAD, DatasetState.QUEUED, DatasetState.RUNNING):
            return True
        return False

[docs]    def ready(self, other_values):
        if value := self._get_dataset_like_object(other_values):
            if value.state == DatasetState.OK:
                return True
            elif value.state == DatasetState.ERROR:
                raise Exception(f"A data display parameter is in the error state: {self.name}")
        return False


[docs]class DisplayApplicationTemplateParameter(DisplayApplicationParameter):
    """Parameter that returns a string containing the requested content"""

    type = "template"

[docs]    def __init__(self, elem, link):
        DisplayApplicationParameter.__init__(self, elem, link)
        self.text = elem.text or ""

[docs]    def get_value(self, other_values, dataset_hash, user_hash, trans):
        value = fill_template(self.text, context=other_values)
        if self.strip:
            value = value.strip()
        return DisplayParameterValueWrapper(value, self, other_values, dataset_hash, user_hash, trans)


parameter_type_to_class = {
    DisplayApplicationDataParameter.type: DisplayApplicationDataParameter,
    DisplayApplicationTemplateParameter.type: DisplayApplicationTemplateParameter,
}


[docs]class DisplayParameterValueWrapper:
    ACTION_NAME = "param"

[docs]    def __init__(self, value, parameter, other_values, dataset_hash, user_hash, trans):
        self.value = value
        self.parameter = parameter
        self.other_values = other_values
        self.trans = trans
        self._dataset_hash = dataset_hash
        self._user_hash = user_hash
        self._url = self.parameter.build_url(self.other_values)

    def __str__(self):
        return str(self.value)

[docs]    def mime_type(self, action_param_extra=None):
        if self.parameter.mime_type is not None:
            return self.parameter.mime_type
        if self.parameter.guess_mime_type:
            mime, encoding = mimetypes.guess_type(self._url)
            if not mime:
                mime = self.trans.app.datatypes_registry.get_mimetype_by_extension(self._url.split(".")[-1], None)
            if mime:
                return mime
        return "text/plain"

    @property
    def url(self):
        base_url = self.trans.request.base
        if self.parameter.strip_https and base_url[:5].lower() == "https":
            base_url = f"http{base_url[5:]}"
        return "{}{}".format(
            base_url,
            self.trans.app.url_for(
                controller="dataset",
                action="display_application",
                dataset_id=self._dataset_hash,
                user_id=self._user_hash,
                app_name=quote_plus(self.parameter.link.display_application.id),
                link_name=quote_plus(self.parameter.link.id),
                app_action=self.action_name,
                action_param=self._url,
            ),
        )

    @property
    def action_name(self):
        return self.ACTION_NAME

    @property
    def qp(self):
        # returns quoted str contents
        return self.other_values["qp"](str(self))

    def __getattr__(self, key):
        return getattr(self.value, key)


[docs]class DisplayDataValueWrapper(DisplayParameterValueWrapper):
    ACTION_NAME = "data"

    def __str__(self):
        # string of data param is filename
        return str(self.value.get_file_name())

[docs]    def mime_type(self, action_param_extra=None):
        if self.parameter.mime_type is not None:
            return self.parameter.mime_type
        if self.parameter.guess_mime_type:
            if action_param_extra:
                mime, encoding = mimetypes.guess_type(action_param_extra)
            else:
                mime, encoding = mimetypes.guess_type(self._url)
            if not mime:
                if action_param_extra:
                    mime = self.trans.app.datatypes_registry.get_mimetype_by_extension(
                        action_param_extra.split(".")[-1], None
                    )
                if not mime:
                    mime = self.trans.app.datatypes_registry.get_mimetype_by_extension(self._url.split(".")[-1], None)
            if mime:
                return mime
        if hasattr(self.value, "get_mime"):
            return self.value.get_mime()
        return self.other_values[DEFAULT_DATASET_NAME].get_mime()

    @property
    def action_name(self):
        if self.parameter.force_url_param:
            return super(DisplayParameterValueWrapper, self).action_name
        return self.ACTION_NAME

    @property
    def qp(self):
        # returns quoted url contents
        return self.other_values["qp"](self.url)