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.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):
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('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('This Galaxy does not recognize the datatype of: %s' % (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' + rval['folder_id']
rval['full_path'] = full_path
rval['file_size'] = util.nice_size(int(ldda.get_size()))
rval['date_uploaded'] = ldda.create_time.strftime("%Y-%m-%d %I:%M %p")
rval['update_time'] = ldda.update_time.strftime("%Y-%m-%d %I:%M %p")
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' + trans.security.encode_id(folder.id), folder.name))
else:
# We add the current folder and traverse up one folder.
path_to_root.append(('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