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.webapps.galaxy.api.library_datasets

"""API operations on the library datasets."""

import glob
import logging
import os
import os.path
import string
from json import dumps

from paste.httpexceptions import (
    HTTPBadRequest,
    HTTPInternalServerError,
)

from galaxy import (
    exceptions,
    util,
    web,
)
from galaxy.actions.library import LibraryActions
from galaxy.exceptions import ObjectNotFound
from galaxy.managers import (
    base as managers_base,
    folders,
    lddas,
    library_datasets,
    roles,
)
from galaxy.model.base import transaction
from galaxy.structured_app import StructuredApp
from galaxy.tools.actions import upload_common
from galaxy.tools.parameters import populate_state
from galaxy.util.path import (
    full_path_permission_for_user,
    safe_contains,
    safe_relpath,
    unsafe_walk,
)
from galaxy.util.zipstream import ZipstreamWrapper
from galaxy.web import (
    expose_api,
    expose_api_anonymous,
)
from galaxy.webapps.base.controller import UsesVisualizationMixin
from . import BaseGalaxyAPIController

log = logging.getLogger(__name__)


[docs]class LibraryDatasetsController(BaseGalaxyAPIController, UsesVisualizationMixin, LibraryActions):
[docs] def __init__(self, app: StructuredApp): super().__init__(app) self.app = app self.folder_manager = folders.FolderManager() self.role_manager = roles.RoleManager(app) self.ld_manager = library_datasets.LibraryDatasetsManager(app) self.ldda_manager = lddas.LDDAManager(app)
[docs] @expose_api_anonymous def show(self, trans, id, **kwd): """ GET /api/libraries/datasets/{encoded_dataset_id} Show the details of a library dataset. :param id: the encoded id of the library dataset to query :type id: an encoded id string :returns: detailed library dataset information :rtype: dictionary """ ld = self.ld_manager.get(trans, managers_base.decode_id(self.app, id)) serialized = self.ld_manager.serialize(trans, ld) return serialized
[docs] @expose_api_anonymous def show_version(self, trans, encoded_dataset_id, encoded_ldda_id, **kwd): """ GET /api/libraries/datasets/{encoded_dataset_id}/versions/{encoded_ldda_id} Display a specific version of a library dataset (i.e. ldda). :param encoded_dataset_id: the encoded id of the related library dataset :type encoded_dataset_id: an encoded id string :param encoded_ldda_id: the encoded id of the ldda to query :type encoded_ldda_id: an encoded id string :returns: dict of ldda's details :rtype: dictionary :raises: ObjectNotFound """ library_dataset = self.ld_manager.get(trans, managers_base.decode_id(self.app, encoded_dataset_id)) try: ldda = self.get_library_dataset_dataset_association( trans, id=encoded_ldda_id, check_ownership=False, check_accessible=False ) except Exception as e: raise exceptions.ObjectNotFound(f"Requested version of library dataset was not found.{util.unicodify(e)}") if ldda not in library_dataset.expired_datasets: raise exceptions.ObjectNotFound("Given library dataset does not have the requested version.") rval = trans.security.encode_all_ids(ldda.to_dict()) return rval
[docs] @expose_api def show_roles(self, trans, encoded_dataset_id, **kwd): """ GET /api/libraries/datasets/{encoded_dataset_id}/permissions Display information about current or available roles for a given dataset permission. :param encoded_dataset_id: the encoded id of the dataset to query :type encoded_dataset_id: an encoded id string :param scope: either 'current' or 'available' :type scope: string :returns: either dict of current roles for all permission types or dict of available roles to choose from (is the same for any permission type) :rtype: dictionary :raises: InsufficientPermissionsException """ current_user_roles = trans.get_current_user_roles() library_dataset = self.ld_manager.get(trans, managers_base.decode_id(self.app, encoded_dataset_id)) dataset = library_dataset.library_dataset_dataset_association.dataset # User has to have manage permissions permission in order to see the roles. can_manage = trans.app.security_agent.can_manage_dataset(current_user_roles, dataset) or trans.user_is_admin if not can_manage: raise exceptions.InsufficientPermissionsException( "You do not have proper permission to access permissions." ) scope = kwd.get("scope", None) if scope in ["current", None]: return self._get_current_roles(trans, library_dataset) elif scope in ["available"]: page = kwd.get("page", None) if page is not None: page = int(page) else: page = 1 page_limit = kwd.get("page_limit", None) if page_limit is not None: page_limit = int(page_limit) else: page_limit = 10 query = kwd.get("q", None) roles, total_roles = trans.app.security_agent.get_valid_roles(trans, dataset, query, page, page_limit) return_roles = [] for role in roles: role_id = trans.security.encode_id(role.id) return_roles.append(dict(id=role_id, name=role.name, type=role.type)) return dict(roles=return_roles, page=page, page_limit=page_limit, total=total_roles) else: raise exceptions.RequestParameterInvalidException( "The value of 'scope' parameter is invalid. Alllowed values: current, available" )
def _get_current_roles(self, trans, library_dataset): """ Find all roles currently connected to relevant permissions on the library dataset and the underlying dataset. :param library_dataset: the model object :type library_dataset: LibraryDataset :rtype: dictionary :returns: dict of current roles for all available permission types """ return self.ldda_manager.serialize_dataset_association_roles(library_dataset)
[docs] @expose_api def update(self, trans, encoded_dataset_id, payload=None, **kwd): """ PATCH /api/libraries/datasets/{encoded_dataset_id} Update the given library dataset (the latest linked ldda). :param encoded_dataset_id: the encoded id of the library dataset to update :type encoded_dataset_id: an encoded id string :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: detailed library dataset information :rtype: dictionary """ library_dataset = self.ld_manager.get(trans, managers_base.decode_id(self.app, encoded_dataset_id)) self.ld_manager.check_modifiable(trans, library_dataset) updated = self.ld_manager.update(library_dataset, payload, trans=trans) serialized = self.ld_manager.serialize(trans, updated) return serialized
[docs] @expose_api def update_permissions(self, trans, encoded_dataset_id, payload=None, **kwd): """ POST /api/libraries/datasets/{encoded_dataset_id}/permissions Set permissions of the given library dataset to the given role ids. :param encoded_dataset_id: the encoded id of the dataset to update permissions of :type encoded_dataset_id: an encoded id string :param payload: dictionary structure containing: :param action: (required) describes what action should be performed available actions: make_private, remove_restrictions, set_permissions :type action: string :param access_ids[]: list of Role.id defining roles that should have access permission on the dataset :type access_ids[]: string or list :param manage_ids[]: list of Role.id defining roles that should have manage permission on the dataset :type manage_ids[]: string or list :param modify_ids[]: list of Role.id defining roles that should have modify permission on the library dataset item :type modify_ids[]: string or list :type: dictionary :returns: dict of current roles for all available permission types :rtype: dictionary :raises: RequestParameterInvalidException, ObjectNotFound, InsufficientPermissionsException, InternalServerError RequestParameterMissingException """ if payload: kwd.update(payload) action = kwd.get("action", None) if action not in ["remove_restrictions", "make_private", "set_permissions"]: raise exceptions.RequestParameterInvalidException( 'The mandatory parameter "action" has an invalid value. ' 'Allowed values are: "remove_restrictions", "make_private", "set_permissions"' ) library_dataset = self.ld_manager.get(trans, managers_base.decode_id(self.app, encoded_dataset_id)) # Some permissions are attached directly to the underlying dataset. dataset = library_dataset.library_dataset_dataset_association.dataset current_user_roles = trans.get_current_user_roles() can_manage = trans.app.security_agent.can_manage_dataset(current_user_roles, dataset) or trans.user_is_admin if not can_manage: raise exceptions.InsufficientPermissionsException( "You do not have proper permissions to manage permissions on this dataset." ) new_access_roles_ids = util.listify(kwd.get("access_ids[]", None)) new_manage_roles_ids = util.listify(kwd.get("manage_ids[]", None)) new_modify_roles_ids = util.listify(kwd.get("modify_ids[]", None)) if action == "remove_restrictions": trans.app.security_agent.make_dataset_public(dataset) if not trans.app.security_agent.dataset_is_public(dataset): raise exceptions.InternalServerError("An error occurred while making dataset public.") elif action == "make_private": if not trans.app.security_agent.dataset_is_private_to_user(trans, dataset): private_role = trans.app.security_agent.get_private_user_role(trans.user) dp = trans.app.model.DatasetPermissions( trans.app.security_agent.permitted_actions.DATASET_ACCESS.action, dataset, private_role ) trans.sa_session.add(dp) with transaction(trans.sa_session): trans.sa_session.commit() if not trans.app.security_agent.dataset_is_private_to_user(trans, dataset): # Check again and inform the user if dataset is not private. raise exceptions.InternalServerError("An error occurred and the dataset is NOT private.") elif action == "set_permissions": # ACCESS DATASET ROLES valid_access_roles = [] invalid_access_roles_ids = [] valid_roles_for_dataset, total_roles = trans.app.security_agent.get_valid_roles(trans, dataset) if new_access_roles_ids is None: trans.app.security_agent.make_dataset_public(dataset) else: for role_id in new_access_roles_ids: role = self.role_manager.get(trans, managers_base.decode_id(self.app, role_id)) if role in valid_roles_for_dataset: valid_access_roles.append(role) else: invalid_access_roles_ids.append(role_id) if len(invalid_access_roles_ids) > 0: log.warning( f"The following roles could not be added to the dataset access permission: {str(invalid_access_roles_ids)}" ) access_permission = dict(access=valid_access_roles) trans.app.security_agent.set_dataset_permission(dataset, access_permission) # MANAGE DATASET ROLES valid_manage_roles = [] invalid_manage_roles_ids = [] new_manage_roles_ids = util.listify(new_manage_roles_ids) for role_id in new_manage_roles_ids: role = self.role_manager.get(trans, managers_base.decode_id(self.app, role_id)) if role in valid_roles_for_dataset: valid_manage_roles.append(role) else: invalid_manage_roles_ids.append(role_id) if len(invalid_manage_roles_ids) > 0: log.warning( f"The following roles could not be added to the dataset manage permission: {str(invalid_manage_roles_ids)}" ) manage_permission = { trans.app.security_agent.permitted_actions.DATASET_MANAGE_PERMISSIONS: valid_manage_roles } trans.app.security_agent.set_dataset_permission(dataset, manage_permission) # MODIFY LIBRARY ITEM ROLES valid_modify_roles = [] invalid_modify_roles_ids = [] new_modify_roles_ids = util.listify(new_modify_roles_ids) for role_id in new_modify_roles_ids: role = self.role_manager.get(trans, managers_base.decode_id(self.app, role_id)) if role in valid_roles_for_dataset: valid_modify_roles.append(role) else: invalid_modify_roles_ids.append(role_id) if len(invalid_modify_roles_ids) > 0: log.warning( f"The following roles could not be added to the dataset modify permission: {str(invalid_modify_roles_ids)}" ) modify_permission = {trans.app.security_agent.permitted_actions.LIBRARY_MODIFY: valid_modify_roles} trans.app.security_agent.set_library_item_permission(library_dataset, modify_permission) return self._get_current_roles(trans, library_dataset)
[docs] @expose_api def delete(self, trans, encoded_dataset_id, **kwd): """ DELETE /api/libraries/datasets/{encoded_dataset_id} Mark the dataset deleted or undeleted. :param encoded_dataset_id: the encoded id of the dataset to change :type encoded_dataset_id: an encoded id string :param undelete: flag whether to undeleted instead of deleting :type undelete: bool :returns: dict containing information about the dataset :rtype: dictionary """ undelete = util.string_as_bool(kwd.get("undelete", False)) library_dataset = self.ld_manager.get(trans, managers_base.decode_id(self.app, encoded_dataset_id)) current_user_roles = trans.get_current_user_roles() allowed = trans.app.security_agent.can_modify_library_item(current_user_roles, library_dataset) if (not allowed) and (not trans.user_is_admin): raise exceptions.InsufficientPermissionsException( "You do not have proper permissions to delete this dataset." ) if undelete: library_dataset.deleted = False else: library_dataset.deleted = True trans.sa_session.add(library_dataset) with transaction(trans.sa_session): trans.sa_session.commit() rval = trans.security.encode_all_ids(library_dataset.to_dict()) nice_size = util.nice_size( int(library_dataset.library_dataset_dataset_association.get_size(calculate_size=False)) ) rval["file_size"] = nice_size rval["update_time"] = library_dataset.update_time.strftime("%Y-%m-%d %I:%M %p") rval["deleted"] = library_dataset.deleted rval["folder_id"] = f"F{rval['folder_id']}" return rval
[docs] @expose_api def load(self, trans, payload=None, **kwd): """ POST /api/libraries/datasets Load dataset(s) from the given source into the library. :param payload: dictionary structure containing: :param encoded_folder_id: the encoded id of the folder to import dataset(s) to :type encoded_folder_id: an encoded id string :param source: source the datasets should be loaded from. Source can be: - user directory root folder specified in galaxy.ini as "$user_library_import_dir" example path: path/to/galaxy/$user_library_import_dir/user@example.com/{user can browse everything here} the folder with the user login has to be created beforehand - (admin)import directory root folder specified in galaxy ini as "$library_import_dir" example path: path/to/galaxy/$library_import_dir/{admin can browse everything here} - (admin)any absolute or relative path option allowed with "allow_library_path_paste" in galaxy.ini :type source: str :param link_data: flag whether to link the dataset to data or copy it to Galaxy, defaults to copy while linking is set to True all symlinks will be resolved _once_ :type link_data: bool :param preserve_dirs: flag whether to preserve the directory structure when importing dir if False only datasets will be imported :type preserve_dirs: bool :param file_type: file type of the loaded datasets, defaults to 'auto' (autodetect) :type file_type: str :param dbkey: dbkey of the loaded genome, defaults to '?' (unknown) :type dbkey: str :param tag_using_filenames: flag whether to generate dataset tags from filenames :type tag_using_filenames: bool :type dictionary :returns: dict containing information about the created upload job :rtype: dictionary :raises: RequestParameterMissingException, AdminRequiredException, ConfigDoesNotAllowException, RequestParameterInvalidException InsufficientPermissionsException, ObjectNotFound """ if payload: kwd.update(payload) kwd["space_to_tab"] = False kwd["to_posix_lines"] = True kwd["dbkey"] = kwd.get("dbkey", "?") kwd["file_type"] = kwd.get("file_type", "auto") kwd["link_data_only"] = "link_to_files" if util.string_as_bool(kwd.get("link_data", False)) else "copy_files" kwd["tag_using_filenames"] = util.string_as_bool(kwd.get("tag_using_filenames", None)) if (encoded_folder_id := kwd.get("encoded_folder_id", None)) is not None: folder_id = self.folder_manager.cut_and_decode(trans, encoded_folder_id) else: raise exceptions.RequestParameterMissingException("The required attribute encoded_folder_id is missing.") path = kwd.get("path", None) if path is None: raise exceptions.RequestParameterMissingException("The required attribute path is missing.") if not isinstance(path, str): raise exceptions.RequestParameterInvalidException("The required attribute path is not String.") folder = self.folder_manager.get(trans, folder_id) source = kwd.get("source", None) if source not in ["userdir_file", "userdir_folder", "importdir_file", "importdir_folder", "admin_path"]: raise exceptions.RequestParameterMissingException( 'You have to specify "source" parameter. Possible values are "userdir_file", "userdir_folder", "admin_path", "importdir_file" and "importdir_folder". ' ) elif source in ["importdir_file", "importdir_folder"]: if not trans.user_is_admin: raise exceptions.AdminRequiredException("Only admins can import from importdir.") if not trans.app.config.library_import_dir: raise exceptions.ConfigDoesNotAllowException( "The configuration of this Galaxy instance does not allow admins to import into library from importdir." ) import_base_dir = trans.app.config.library_import_dir if not safe_relpath(path): # admins shouldn't be able to explicitly specify a path outside server_dir, but symlinks are allowed. # the reasoning here is that galaxy admins may not have direct filesystem access or can only access # library_import_dir via FTP (which cannot create symlinks), and may rely on sysadmins to set up the # import directory. if they have filesystem access, all bets are off. raise exceptions.RequestParameterInvalidException("The given path is invalid.") path = os.path.join(import_base_dir, path) elif source in ["userdir_file", "userdir_folder"]: username = trans.user.username if trans.app.config.user_library_import_check_permissions else None user_login = trans.user.email user_base_dir = trans.app.config.user_library_import_dir if user_base_dir is None: raise exceptions.ConfigDoesNotAllowException( "The configuration of this Galaxy instance does not allow upload from user directories." ) full_dir = os.path.join(user_base_dir, user_login) if not safe_contains(full_dir, path, allowlist=trans.app.config.user_library_import_symlink_allowlist): # the path is a symlink outside the user dir path = os.path.join(full_dir, path) log.error( "User attempted to import a path that resolves to a path outside of their import dir: %s -> %s", path, os.path.realpath(path), ) raise exceptions.RequestParameterInvalidException("The given path is invalid.") if username is not None and not full_path_permission_for_user(full_dir, path, username): log.error( "User attempted to import a path that resolves to a path outside of their import dir: " "%s -> %s and cannot be read by them.", path, os.path.realpath(path), ) raise exceptions.RequestParameterInvalidException("The given path is invalid.") path = os.path.join(full_dir, path) if unsafe_walk( path, allowlist=[full_dir] + trans.app.config.user_library_import_symlink_allowlist, username=username ): # the path is a dir and contains files that symlink outside the user dir error = f"User attempted to import a path that resolves to a path outside of their import dir: {path} -> {os.path.realpath(path)}" if trans.app.config.user_library_import_check_permissions: error += " or is not readable for them." log.error(error) raise exceptions.RequestParameterInvalidException("The given path is invalid.") if not os.path.exists(path): raise exceptions.RequestParameterInvalidException("Given path does not exist on the host.") if not self.folder_manager.can_add_item(trans, folder): raise exceptions.InsufficientPermissionsException( "You do not have proper permission to add items to the given folder." ) elif source == "admin_path": if not trans.app.config.allow_library_path_paste: raise exceptions.ConfigDoesNotAllowException( "The configuration of this Galaxy instance does not allow admins to import into library from path." ) if not trans.user_is_admin: raise exceptions.AdminRequiredException("Only admins can import from path.") # Set up the traditional tool state/params tool_id = "upload1" tool = trans.app.toolbox.get_tool(tool_id) state = tool.new_state(trans) populate_state(trans, tool.inputs, kwd, state.inputs) tool_params = state.inputs dataset_upload_inputs = [] for input in tool.inputs.values(): if input.type == "upload_dataset": dataset_upload_inputs.append(input) library_bunch = upload_common.handle_library_params(trans, {}, folder.id) abspath_datasets = [] kwd["filesystem_paths"] = path if source in ["importdir_folder"]: kwd["filesystem_paths"] = os.path.join(import_base_dir, path) # user wants to import one file only elif source in ["userdir_file", "importdir_file"]: file = os.path.abspath(path) abspath_datasets.append( self._make_library_uploaded_dataset( trans, kwd, os.path.basename(file), file, "server_dir", library_bunch ) ) # user wants to import whole folder elif source == "userdir_folder": uploaded_datasets_bunch = self._get_path_paste_uploaded_datasets(trans, kwd, library_bunch, 200, "") uploaded_datasets = uploaded_datasets_bunch[0] if uploaded_datasets is None: raise exceptions.ObjectNotFound("Given folder does not contain any datasets.") for ud in uploaded_datasets: ud.path = os.path.abspath(ud.path) abspath_datasets.append(ud) # user wants to import from path if source in ["admin_path", "importdir_folder"]: # validate the path is within root uploaded_datasets_bunch = self._get_path_paste_uploaded_datasets(trans, kwd, library_bunch, 200, "") uploaded_datasets = uploaded_datasets_bunch[0] if uploaded_datasets is None: raise exceptions.ObjectNotFound("Given folder does not contain any datasets.") for ud in uploaded_datasets: ud.path = os.path.abspath(ud.path) abspath_datasets.append(ud) json_file_path = upload_common.create_paramfile(trans, abspath_datasets) data_list = [ud.data for ud in abspath_datasets] job_params = {} job_params["link_data_only"] = dumps(kwd.get("link_data_only", "copy_files")) job_params["uuid"] = dumps(kwd.get("uuid", None)) job, output = upload_common.create_job( trans, tool_params, tool, json_file_path, data_list, folder=folder, job_params=job_params ) trans.app.job_manager.enqueue(job, tool=tool) job_dict = job.to_dict() job_dict["id"] = trans.security.encode_id(job_dict["id"]) return job_dict
[docs] @web.expose # TODO convert to expose_api def download(self, trans, archive_format, **kwd): """ GET /api/libraries/datasets/download/{archive_format} POST /api/libraries/datasets/download/{archive_format} Download requested datasets (identified by encoded IDs) in requested archive_format. example: ``GET localhost:8080/api/libraries/datasets/download/tbz?ld_ids%255B%255D=a0d84b45643a2678&ld_ids%255B%255D=fe38c84dcd46c828`` .. note:: supported archive_format values are: 'zip', 'tgz', 'tbz', 'uncompressed' :param archive_format: string representing requested archive archive_format :type archive_format: string :param ld_ids[]: an array of encoded dataset ids :type ld_ids[]: an array :param folder_ids[]: an array of encoded folder ids :type folder_ids[]: an array :returns: either archive with the requested datasets packed inside or a single uncompressed dataset :rtype: file :raises: MessageException, ItemDeletionException, ItemAccessibilityException, HTTPBadRequest, OSError, IOError, ObjectNotFound """ library_datasets = [] datasets_to_download = kwd.get("ld_ids%5B%5D", None) if datasets_to_download is None: datasets_to_download = kwd.get("ld_ids", None) if datasets_to_download is not None: datasets_to_download = util.listify(datasets_to_download) for dataset_id in datasets_to_download: try: library_dataset = self.get_library_dataset( trans, id=dataset_id, check_ownership=False, check_accessible=True ) library_datasets.append(library_dataset) except HTTPBadRequest: raise exceptions.RequestParameterInvalidException("Bad Request.") except HTTPInternalServerError: raise exceptions.InternalServerError("Internal error.") except Exception as e: raise exceptions.InternalServerError(f"Unknown error.{util.unicodify(e)}") folders_to_download = kwd.get("folder_ids%5B%5D", None) if folders_to_download is None: folders_to_download = kwd.get("folder_ids", None) if folders_to_download is not None: folders_to_download = util.listify(folders_to_download) current_user_roles = trans.get_current_user_roles() def traverse(folder): admin = trans.user_is_admin rval = [] for subfolder in folder.active_folders: if not admin: can_access, folder_ids = trans.app.security_agent.check_folder_contents( trans.user, current_user_roles, subfolder ) if (admin or can_access) and not subfolder.deleted: rval.extend(traverse(subfolder)) for ld in folder.datasets: if not admin: can_access = trans.app.security_agent.can_access_dataset( current_user_roles, ld.library_dataset_dataset_association.dataset ) if (admin or can_access) and not ld.deleted: rval.append(ld) return rval for encoded_folder_id in folders_to_download: folder_id = self.folder_manager.cut_and_decode(trans, encoded_folder_id) folder = self.folder_manager.get(trans, folder_id) library_datasets.extend(traverse(folder)) if not library_datasets: raise exceptions.RequestParameterMissingException( "Request has to contain a list of dataset ids or folder ids to download." ) if archive_format == "zip": archive = ZipstreamWrapper( archive_name="selected_library_files", upstream_mod_zip=self.app.config.upstream_mod_zip, upstream_gzip=self.app.config.upstream_gzip, ) killme = string.punctuation + string.whitespace trantab = str.maketrans(killme, "_" * len(killme)) seen = [] for ld in library_datasets: ldda = ld.library_dataset_dataset_association is_composite = ldda.datatype.composite_type path = "" parent_folder = ldda.library_dataset.folder while parent_folder is not None: # Exclude the now-hidden "root folder" if parent_folder.parent is None: path = os.path.join(parent_folder.library_root[0].name, path) break path = os.path.join(parent_folder.name, path) parent_folder = parent_folder.parent path += ldda.name while path in seen: path += "_" path = f"{path}.{ldda.extension}" seen.append(path) zpath = os.path.split(path)[-1] # comes as base_name/fname outfname, zpathext = os.path.splitext(zpath) if is_composite: # need to add all the components from the extra_files_path to the zip if zpathext == "": zpath = f"{zpath}.html" # fake the real nature of the html file try: if archive_format == "zip": archive.write(ldda.dataset.get_file_name(), zpath) # add the primary of a composite set else: archive.write(ldda.dataset.get_file_name(), zpath) # add the primary of a composite set except OSError: log.exception( "Unable to add composite parent %s to temporary library download archive", ldda.dataset.get_file_name(), ) raise exceptions.InternalServerError("Unable to create archive for download.") except ObjectNotFound: log.exception("Requested dataset %s does not exist on the host.", ldda.dataset.get_file_name()) raise exceptions.ObjectNotFound("Requested dataset not found. ") except Exception as e: log.exception( "Unable to add composite parent %s to temporary library download archive", ldda.dataset.get_file_name(), ) raise exceptions.InternalServerError( f"Unable to add composite parent to temporary library download archive. {util.unicodify(e)}" ) flist = glob.glob(os.path.join(ldda.dataset.extra_files_path, "*.*")) # glob returns full paths for fpath in flist: efp, fname = os.path.split(fpath) if fname > "": fname = fname.translate(trantab) try: archive.write(fpath, fname) except OSError: log.exception("Unable to add %s to temporary library download archive %s", fname, outfname) raise exceptions.InternalServerError("Unable to create archive for download.") except ObjectNotFound: log.exception("Requested dataset %s does not exist on the host.", fpath) raise exceptions.ObjectNotFound("Requested dataset not found.") except Exception as e: log.exception("Unable to add %s to temporary library download archive %s", fname, outfname) raise exceptions.InternalServerError( f"Unable to add dataset to temporary library download archive . {util.unicodify(e)}" ) else: try: archive.write(ldda.dataset.get_file_name(), path) except OSError: log.exception( "Unable to write %s to temporary library download archive", ldda.dataset.get_file_name() ) raise exceptions.InternalServerError("Unable to create archive for download") except ObjectNotFound: log.exception("Requested dataset %s does not exist on the host.", ldda.dataset.get_file_name()) raise exceptions.ObjectNotFound("Requested dataset not found.") except Exception as e: log.exception( "Unable to add %s to temporary library download archive %s", ldda.dataset.get_file_name(), outfname, ) raise exceptions.InternalServerError(f"Unknown error. {util.unicodify(e)}") trans.response.headers.update(archive.get_headers()) return archive.response() elif archive_format == "uncompressed": if len(library_datasets) != 1: raise exceptions.RequestParameterInvalidException( "You can download only one uncompressed file at once." ) else: single_ld = library_datasets[0] ldda = single_ld.library_dataset_dataset_association dataset = ldda.dataset fStat = os.stat(dataset.get_file_name()) trans.response.set_content_type(ldda.get_mime()) trans.response.headers["Content-Length"] = str(fStat.st_size) fname = f"{ldda.name}.{ldda.extension}" fname = "".join(c in util.FILENAME_VALID_CHARS and c or "_" for c in fname)[0:150] trans.response.headers["Content-Disposition"] = f'attachment; filename="{fname}"' try: return open(dataset.get_file_name(), "rb") except Exception: raise exceptions.InternalServerError("This dataset contains no content.") else: raise exceptions.RequestParameterInvalidException("Wrong archive_format parameter specified")