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.remote_files

import hashlib
import logging
from operator import itemgetter
from typing import (

from galaxy import exceptions
from galaxy.files import (
from galaxy.managers.context import ProvidesUserContext
from galaxy.schema.remote_files import (
from galaxy.structured_app import MinimalManagerApp
from galaxy.util import (

log = logging.getLogger(__name__)

[docs]class RemoteFilesManager: """ Interface/service object for interacting with remote files. """
[docs] def __init__(self, app: MinimalManagerApp): self._app = app
[docs] def index( self, user_ctx: ProvidesUserContext, target: str, format: Optional[RemoteFilesFormat], recursive: Optional[bool], disable: Optional[RemoteFilesDisableMode], ) -> List[Dict[str, Any]]: """Returns a list of remote files available to the user.""" user_file_source_context = ProvidesUserFileSourcesUserContext(user_ctx) default_recursive = False default_format = RemoteFilesFormat.uri if "://" in target: uri = target elif target == RemoteFilesTarget.userdir: uri = "gxuserimport://" default_format = RemoteFilesFormat.flat default_recursive = True elif target == RemoteFilesTarget.importdir: uri = "gximport://" default_format = RemoteFilesFormat.flat default_recursive = True elif target in [RemoteFilesTarget.ftpdir, "ftp"]: # legacy, allow both uri = "gxftp://" default_format = RemoteFilesFormat.flat default_recursive = True else: raise exceptions.RequestParameterInvalidException(f"Invalid target parameter supplied [{target}]") if format is None: format = default_format if recursive is None: recursive = default_recursive self._file_sources.validate_uri_root(uri, user_context=user_file_source_context) file_source_path = self._file_sources.get_file_source_path(uri) file_source = file_source_path.file_source try: index = file_source.list(file_source_path.path, recursive=recursive, user_context=user_file_source_context) except exceptions.MessageException: log.warning(f"Problem listing file source path {file_source_path}", exc_info=True) raise except Exception: message = f"Problem listing file source path {file_source_path}" log.warning(message, exc_info=True) raise exceptions.InternalServerError(message) if format == RemoteFilesFormat.flat: # rip out directories, ensure sorted by path index = [i for i in index if i["class"] == "File"] index = sorted(index, key=itemgetter("path")) if format == RemoteFilesFormat.jstree: if disable is None: disable = RemoteFilesDisableMode.folders jstree_paths = [] for ent in index: path = ent["path"] path_hash = hashlib.sha1(smart_str(path)).hexdigest() if ent["class"] == "Directory": path_type = "folder" disabled = True if disable == RemoteFilesDisableMode.folders else False else: path_type = "file" disabled = True if disable == RemoteFilesDisableMode.files else False jstree_paths.append( jstree.Path( path, path_hash, {"type": path_type, "state": {"disabled": disabled}, "li_attr": {"full_path": path}}, ) ) userdir_jstree = jstree.JSTree(jstree_paths) index = userdir_jstree.jsonData() return index
[docs] def get_files_source_plugins(self, user_context: ProvidesUserContext) -> FilesSourcePluginList: """Display plugin information for each of the gxfiles:// URI targets available.""" user_file_source_context = ProvidesUserFileSourcesUserContext(user_context) plugins = self._file_sources.plugins_to_dict(user_context=user_file_source_context) return FilesSourcePluginList.parse_obj(plugins)
@property def _file_sources(self) -> ConfiguredFileSources: return self._app.file_sources