"""Manager and Serializer for library datasets."""
import logging
from sqlalchemy import select
from galaxy import (
model,
util,
)
from galaxy.exceptions import (
InsufficientPermissionsException,
InternalServerError,
ObjectNotFound,
RequestParameterInvalidException,
)
from galaxy.managers import datasets
from galaxy.managers.context import ProvidesUserContext
from galaxy.model import (
LibraryDataset,
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(datasets.DatasetAssociationManager):
"""Interface/service object for interacting with library datasets."""
model_class = model.LibraryDatasetDatasetAssociation
[docs] def __init__(self, app: MinimalManagerApp):
self.app = app
[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 = 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, 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: ProvidesUserContext, 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:
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()
trans.sa_session.add(ldda)
with transaction(trans.sa_session):
trans.sa_session.commit()
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"] = 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 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.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):
stmt = select(LibraryDataset).where(LibraryDataset.id == library_dataset_id)
return session.scalars(stmt).one()