Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.managers.library_datasets

"""Manager and Serializer for library datasets."""
import logging

from galaxy import (
    model,
    util,
)
from galaxy.exceptions import (
    InsufficientPermissionsException,
    InternalServerError,
    ObjectNotFound,
    RequestParameterInvalidException,
)
from galaxy.managers import datasets
from galaxy.model import tags
from galaxy.structured_app import MinimalManagerApp
from galaxy.util import validation

log = logging.getLogger(__name__)


[docs]class LibraryDatasetsManager(datasets.DatasetAssociationManager): """Interface/service object for interacting with library datasets.""" model_class = model.LibraryDatasetDatasetAssociation
[docs] def __init__(self, app: MinimalManagerApp): self.app = app self.tag_handler = tags.GalaxyTagHandler(app.model.context)
[docs] def get(self, trans, decoded_library_dataset_id, check_accessible=True): """ Get the library dataset from the DB. :param decoded_library_dataset_id: decoded library dataset id :type decoded_library_dataset_id: int :param check_accessible: flag whether to check that user can access item :type check_accessible: bool :returns: the requested library dataset :rtype: galaxy.model.LibraryDataset """ try: ld = ( trans.sa_session.query(trans.app.model.LibraryDataset) .filter(trans.app.model.LibraryDataset.table.c.id == decoded_library_dataset_id) .one() ) except Exception as e: raise InternalServerError(f"Error loading from the database.{util.unicodify(e)}") ld = self.secure(trans, ld, check_accessible) return ld
[docs] def update(self, trans, ld, payload): """ Update the given library dataset - the latest linked ldda. Updating older lddas (versions) is not allowed. :param ld: library dataset to change :type ld: LibraryDataset :param payload: dictionary structure containing:: :param name: new ld's name, must be longer than 0 :type name: str :param misc_info: new ld's misc info :type misc_info: str :param file_ext: new ld's extension, must exist in the Galaxy registry :type file_ext: str :param genome_build: new ld's genome build :type genome_build: str :param tags: list of dataset tags :type tags: list :type payload: dict :returns: the changed library dataset :rtype: galaxy.model.LibraryDataset """ self.check_modifiable(trans, ld) # we are going to operate on the actual latest ldda ldda = ld.library_dataset_dataset_association payload = self._validate_and_parse_update_payload(payload) self._set_from_dict(trans, ldda, payload) return ld
def _set_from_dict(self, trans, ldda, new_data): changed = False new_name = new_data.get("name", None) if new_name is not None and new_name != ldda.name: ldda.name = new_name changed = True new_misc_info = new_data.get("misc_info", None) if new_misc_info is not None and new_misc_info != ldda.info: ldda.info = new_misc_info changed = True new_message = new_data.get("message", None) if new_message is not None and new_message != ldda.message: ldda.message = new_message changed = True new_file_ext = new_data.get("file_ext", None) if new_file_ext == "auto": self.detect_datatype(trans, ldda) elif new_file_ext is not None and new_file_ext != ldda.extension: ldda.extension = new_file_ext self.set_metadata(trans, ldda) changed = True new_genome_build = new_data.get("genome_build", None) if new_genome_build is not None and new_genome_build != ldda.dbkey: ldda.dbkey = new_genome_build changed = True new_tags = new_data.get("tags", None) if new_tags is not None and new_tags != ldda.tags: self.tag_handler.delete_item_tags(item=ldda, user=trans.user) tag_list = self.tag_handler.parse_tags_list(new_tags) for tag in tag_list: self.tag_handler.apply_item_tag(item=ldda, user=trans.user, name=tag[0], value=tag[1]) changed = True if changed: ldda.update_parent_folder_update_times() trans.sa_session.add(ldda) trans.sa_session.flush() return changed def _validate_and_parse_update_payload(self, payload): MINIMUM_STRING_LENGTH = 1 validated_payload = {} for key, val in payload.items(): if val is None: continue if key in ("name"): if len(val) < MINIMUM_STRING_LENGTH: raise RequestParameterInvalidException( f"{key} must have at least length of {MINIMUM_STRING_LENGTH}" ) val = validation.validate_and_sanitize_basestring(key, val) validated_payload[key] = val if key in ("misc_info", "message"): val = validation.validate_and_sanitize_basestring(key, val) validated_payload[key] = val if key in ("file_ext"): datatype = self.app.datatypes_registry.get_datatype_by_extension(val) if datatype is None and val not in ("auto",): raise RequestParameterInvalidException(f"This Galaxy does not recognize the datatype of: {val}") validated_payload[key] = val if key in ("genome_build"): if len(val) < MINIMUM_STRING_LENGTH: raise RequestParameterInvalidException( f"{key} must have at least length of {MINIMUM_STRING_LENGTH}" ) val = validation.validate_and_sanitize_basestring(key, val) validated_payload[key] = val if key in ("tags"): val = validation.validate_and_sanitize_basestring_list(key, util.listify(val)) validated_payload[key] = val return validated_payload
[docs] def secure(self, trans, ld, check_accessible=True, check_ownership=False): """ Check if library dataset is accessible to current user or the user is an admin. :param ld: library dataset :type ld: galaxy.model.LibraryDataset :param check_accessible: flag whether to check that user can access library dataset :type check_accessible: bool :returns: the original library dataset :rtype: galaxy.model.LibraryDataset """ if trans.user_is_admin: # all operations are available to an admin return ld if check_accessible: ld = self.check_accessible(trans, ld) return ld
[docs] def check_accessible(self, trans, ld): """ Check whether the current user has permissions to access library dataset. :param ld: library dataset :type ld: galaxy.model.LibraryDataset :returns: the original library dataset :rtype: galaxy.model.LibraryDataset :raises: ObjectNotFound """ if not trans.app.security_agent.can_access_library_item(trans.get_current_user_roles(), ld, trans.user): raise ObjectNotFound("Library dataset with the id provided was not found.") elif ld.deleted: raise ObjectNotFound("Library dataset with the id provided is deleted.") else: return ld
[docs] def check_modifiable(self, trans, ld): """ Check whether the current user has permissions to modify library dataset. :param ld: library dataset :type ld: galaxy.model.LibraryDataset :returns: the original library dataset :rtype: galaxy.model.LibraryDataset :raises: ObjectNotFound """ if ld.deleted: raise ObjectNotFound("Library dataset with the id provided is deleted.") elif trans.user_is_admin: return ld if not trans.app.security_agent.can_modify_library_item(trans.get_current_user_roles(), ld): raise InsufficientPermissionsException("You do not have proper permission to modify this library dataset.") else: return ld
[docs] def serialize(self, trans, ld): """Serialize the library dataset into a dictionary.""" current_user_roles = trans.get_current_user_roles() # Build the full path for breadcrumb purposes. full_path = self._build_path(trans, ld.folder) dataset_item = (trans.security.encode_id(ld.id), ld.name) full_path.insert(0, dataset_item) full_path = full_path[::-1] # Find expired versions of the library dataset expired_ldda_versions = [] for expired_ldda in ld.expired_datasets: expired_ldda_versions.append((trans.security.encode_id(expired_ldda.id), expired_ldda.name)) rval = trans.security.encode_all_ids(ld.to_dict()) if len(expired_ldda_versions) > 0: rval["has_versions"] = True rval["expired_versions"] = expired_ldda_versions ldda = ld.library_dataset_dataset_association if ldda.creating_job_associations: if ldda.creating_job_associations[0].job.stdout: rval["job_stdout"] = ldda.creating_job_associations[0].job.stdout.strip() if ldda.creating_job_associations[0].job.stderr: rval["job_stderr"] = ldda.creating_job_associations[0].job.stderr.strip() if ldda.dataset.uuid: rval["uuid"] = str(ldda.dataset.uuid) rval["deleted"] = ld.deleted rval["folder_id"] = f"F{rval['folder_id']}" rval["full_path"] = full_path rval["file_size"] = util.nice_size(int(ldda.get_size(calculate_size=False))) rval["date_uploaded"] = ldda.create_time.isoformat() rval["update_time"] = ldda.update_time.isoformat() rval["can_user_modify"] = trans.user_is_admin or trans.app.security_agent.can_modify_library_item( current_user_roles, ld ) rval["is_unrestricted"] = trans.app.security_agent.dataset_is_public(ldda.dataset) rval["tags"] = self.tag_handler.get_tags_str(ldda.tags) # Manage dataset permission is always attached to the dataset itself, not the the ld or ldda to maintain consistency rval["can_user_manage"] = trans.user_is_admin or trans.app.security_agent.can_manage_dataset( current_user_roles, ldda.dataset ) return rval
def _build_path(self, trans, folder): """ Search the path upwards recursively and load the whole route of names and ids for breadcrumb building purposes. :param folder: current folder for navigating up :param type: Galaxy LibraryFolder :returns: list consisting of full path to the library :type: list """ path_to_root = [] if folder.parent_id is None: # We are almost in root path_to_root.append((f"F{trans.security.encode_id(folder.id)}", folder.name)) else: # We add the current folder and traverse up one folder. path_to_root.append((f"F{trans.security.encode_id(folder.id)}", folder.name)) upper_folder = trans.sa_session.query(trans.app.model.LibraryFolder).get(folder.parent_id) path_to_root.extend(self._build_path(trans, upper_folder)) return path_to_root