Warning

This document is for an in-development version 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 typing import (
    Any,
    Dict,
)

from sqlalchemy import select

from galaxy import util
from galaxy.exceptions import (
    InsufficientPermissionsException,
    InternalServerError,
    ObjectNotFound,
    RequestParameterInvalidException,
)
from galaxy.managers.base import ModelManager
from galaxy.managers.context import ProvidesUserContext
from galaxy.managers.datasets import DatasetAssociationManager
from galaxy.model import (
    LibraryDataset,
    LibraryDatasetDatasetAssociation,
    LibraryFolder,
)
from galaxy.model.base import transaction
from galaxy.structured_app import MinimalManagerApp
from galaxy.util import validation

log = logging.getLogger(__name__)


[docs]class LibraryDatasetsManager(ModelManager[LibraryDataset]): """Interface/service object for interacting with library datasets.""" model_class = LibraryDataset
[docs] def __init__(self, app: MinimalManagerApp): super().__init__(app) self.dataset_assoc_manager = DatasetAssociationManager(app)
[docs] def get(self, trans, decoded_library_dataset_id, check_accessible=True) -> LibraryDataset: """ 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 = get_library_dataset(trans.sa_session, decoded_library_dataset_id) 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, item: LibraryDataset, new_values: Dict[str, Any], flush: bool = True, **kwargs) -> LibraryDataset: """ Update the given library dataset - the latest linked ldda. Updating older lddas (versions) is not allowed. :param item: library dataset to change :type item: LibraryDataset :param new_values: 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 new_values: dict :returns: the changed library dataset :rtype: galaxy.model.LibraryDataset """ trans = kwargs.get("trans") if not trans: raise ValueError("Missing trans parameter") # we are going to operate on the actual latest ldda ldda = item.library_dataset_dataset_association new_values = self._validate_and_parse_update_payload(new_values) self._set_from_dict(trans, ldda, new_values, flush=flush) return item
def _set_from_dict( self, trans: ProvidesUserContext, ldda: LibraryDatasetDatasetAssociation, new_data: Dict[str, Any], flush: bool = True, ) -> None: 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.dataset_assoc_manager.detect_datatype(trans, ldda) elif new_file_ext is not None and new_file_ext != ldda.extension: ldda.extension = new_file_ext self.dataset_assoc_manager.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: trans.tag_handler.delete_item_tags(item=ldda, user=trans.user) tag_list = trans.tag_handler.parse_tags_list(new_tags) for tag in tag_list: trans.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() session = self.session() session.add(ldda) if flush: with transaction(session): session.commit() 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 self.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 self.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: LibraryDataset) -> Dict[str, Any]: """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 = (self.app.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((self.app.security.encode_id(expired_ldda.id), expired_ldda.name)) rval = self.app.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 self.app.security_agent.can_modify_library_item( current_user_roles, ld ) rval["is_unrestricted"] = self.app.security_agent.dataset_is_public(ldda.dataset) rval["tags"] = trans.tag_handler.get_tags_list(ldda.tags) # Manage dataset permission is always attached to the dataset itself, not the ld or ldda to maintain consistency rval["can_user_manage"] = trans.user_is_admin or self.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{self.app.security.encode_id(folder.id)}", folder.name)) else: # We add the current folder and traverse up one folder. path_to_root.append((f"F{self.app.security.encode_id(folder.id)}", folder.name)) upper_folder = trans.sa_session.get(LibraryFolder, folder.parent_id) path_to_root.extend(self._build_path(trans, upper_folder)) return path_to_root
[docs]def get_library_dataset(session, library_dataset_id) -> LibraryDataset: stmt = select(LibraryDataset).where(LibraryDataset.id == library_dataset_id) return session.scalars(stmt).one()