Source code for galaxy.visualization.plugins.resource_parser

"""
Deserialize Galaxy resources (hdas, ldas, datasets, genomes, etc.) from
a dictionary of string data/ids (often from a query string).
"""

import json
import logging
import weakref
from typing import (
    Callable,
    Dict,
    Optional,
    Union,
)

import galaxy.exceptions
import galaxy.util
from galaxy.managers import (
    hdas as hda_manager,
    visualizations as visualization_manager,
)
from galaxy.model import (
    HistoryDatasetAssociation,
    LibraryDatasetDatasetAssociation,
    Visualization,
)
from galaxy.util import bunch

log = logging.getLogger(__name__)


ParameterPrimitiveType = Union[int, float, str]
ParameterType = Union[
    ParameterPrimitiveType, HistoryDatasetAssociation, LibraryDatasetDatasetAssociation, Visualization
]


[docs]class ResourceParser: """ Given a parameter dictionary (often a converted query string) and a configuration dictionary (curr. only VisualizationsRegistry uses this), convert the entries in the parameter dictionary into resources (Galaxy models, primitive types, lists of either, etc.) and return in a new dictionary. The keys used to store the new values can optionally be re-mapped to new keys (e.g. dataset_id="NNN" -> hda=<HistoryDatasetAssociation>). """ primitive_parsers: Dict[str, Callable[[str], ParameterPrimitiveType]] = { "str": lambda param: galaxy.util.sanitize_html.sanitize_html(param), "bool": lambda param: galaxy.util.string_as_bool(param), "int": int, "float": float, # 'date' : lambda param: , "json": (lambda param: json.loads(galaxy.util.sanitize_html.sanitize_html(param))), }
[docs] def __init__(self, app, *args, **kwargs): self.app = weakref.ref(app) self.managers = self._init_managers(app)
def _init_managers(self, app): return bunch.Bunch( visualization=app[visualization_manager.VisualizationManager], hda=app[hda_manager.HDAManager], )
[docs] def parse_parameter_dictionary(self, trans, param_config_dict, query_params, param_modifiers=None): """ Parse all expected params from the query dictionary `query_params`. If param is required and not present, raises a `KeyError`. """ # log.debug( 'parse_parameter_dictionary, query_params:\n%s', query_params ) # parse the modifiers first since they modify the params coming next # TODO: this is all really for hda_ldda - which we could replace with model polymorphism params_that_modify_other_params = self.parse_parameter_modifiers(trans, param_modifiers, query_params) resources = {} for param_name, param_config in param_config_dict.items(): # optionally rename the variable returned, defaulting to the original name var_name_in_template = param_config.get("var_name_in_template", param_name) # if the param is present, get its value, any param modifiers for that param, and parse it into a resource # use try catch here and not caller to fall back on the default value or re-raise if required resource = None query_val = query_params.get(param_name, None) if query_val is not None: try: target_param_modifiers = params_that_modify_other_params.get(param_name, None) resource = self.parse_parameter( trans, param_config, query_val, param_modifiers=target_param_modifiers ) except Exception as exception: if trans.debug: raise else: log.warning( "Exception parsing visualization param from query: %s, %s, (%s) %s", param_name, query_val, str(type(exception)), str(exception), ) resource = None # here - we've either had no value in the query_params or there was a failure to parse # so: error if required, otherwise get a default (which itself defaults to None) if resource is None: if param_config["required"]: raise KeyError(f"required param {param_name} not found in URL") resource = self.parse_parameter_default(trans, param_config) resources[var_name_in_template] = resource return resources
[docs] def parse_config(self, trans, param_config_dict, query_params): """ Return `query_params` dict parsing only JSON serializable params. Complex params such as models, etc. are left as the original query value. Keys in `query_params` not found in the `param_config_dict` will not be returned. """ # log.debug( 'parse_config, query_params:\n%s', query_params ) config = {} for param_name, param_config in param_config_dict.items(): config_val = query_params.get(param_name, None) if config_val is not None and param_config["type"] in self.primitive_parsers: try: config_val = self.parse_parameter(trans, param_config, config_val) except Exception as exception: log.warning( "Exception parsing visualization param from query: %s, %s, (%s) %s", param_name, config_val, type(exception), exception, ) config_val = None # here - we've either had no value in the query_params or there was a failure to parse # so: if there's a default and it's not None, add it to the config if config_val is None: if param_config.get("default", None) is None: continue config_val = self.parse_parameter_default(trans, param_config) config[param_name] = config_val return config
# TODO: I would LOVE to rip modifiers out completely
[docs] def parse_parameter_modifiers( self, trans, param_modifiers, query_params ) -> Dict[str, Dict[str, Optional[ParameterType]]]: """ Parse and return parameters that are meant to modify other parameters, be grouped with them, or are needed to successfully parse other parameters. """ # only one level of modification - down that road lies madness # parse the modifiers out of query_params first since they modify the other params coming next parsed_modifiers: Dict[str, Dict[str, Optional[ParameterType]]] = {} if not param_modifiers: return parsed_modifiers # precondition: expects a two level dictionary # { target_param_name -> { param_modifier_name -> { param_modifier_data }}} for target_param_name, modifier_dict in param_modifiers.items(): target_modifiers: Dict[str, Optional[ParameterType]] = {} parsed_modifiers[target_param_name] = target_modifiers for modifier_name, modifier_config in modifier_dict.items(): query_val = query_params.get(modifier_name, None) if query_val is not None: modifier = self.parse_parameter(trans, modifier_config, query_val) target_modifiers[modifier_name] = modifier else: # TODO: required attr? target_modifiers[modifier_name] = self.parse_parameter_default(trans, modifier_config) return parsed_modifiers
[docs] def parse_parameter_default(self, trans, param_config) -> Optional[ParameterType]: """ Parse any default values for the given param, defaulting the default to `None`. """ # currently, *default* default is None, so this is quaranteed to be part of the dictionary default = param_config["default"] # if default is None, do not attempt to parse it if default is None: return default # otherwise, parse (currently param_config['default'] is a string just like query param and needs to be parsed) # this saves us the trouble of parsing the default when the config file is read # (and adding this code to the xml parser) return self.parse_parameter(trans, param_config, default)
[docs] def parse_parameter(self, trans, expected_param_data, query_param, recurse=True, param_modifiers=None): """ Use data in `expected_param_data` to parse `query_param` from a string into a resource usable directly by a template. """ param_type = expected_param_data.get("type") parsed_param: Optional[ParameterType] = None if param_type in self.primitive_parsers: # TODO: what about param modifiers on primitives? parsed_param = self.primitive_parsers[param_type](query_param) # TODO: constrain_to: this gets complicated - remove? # db models elif param_type == "visualization": # ?: is this even used anymore/anywhere? decoded_visualization_id = trans.security.decode_id(query_param, object_name=param_type) parsed_param = self.managers.visualization.get_accessible(decoded_visualization_id, trans.user) elif param_type == "dataset": decoded_dataset_id = trans.security.decode_id(query_param, object_name=param_type) parsed_param = self.managers.hda.get_accessible(decoded_dataset_id, trans.user) elif param_type == "hda_or_ldda": encoded_dataset_id = query_param # needs info from another param... hda_ldda = param_modifiers.get("hda_ldda") if hda_ldda == "hda": decoded_dataset_id = trans.security.decode_id(query_param, object_name="dataset") parsed_param = self.managers.hda.get_accessible(decoded_dataset_id, trans.user) else: parsed_param = self.managers.ldda.get(trans, encoded_dataset_id) # TODO: ideally this would check v. a list of valid dbkeys elif param_type == "dbkey": dbkey = query_param parsed_param = galaxy.util.sanitize_html.sanitize_html(dbkey) return parsed_param