Source code for tool_shed.webapp.controllers.upload

import logging
import os
import shutil
import tarfile
import tempfile

import requests

from galaxy import (
    util,
    web,
)
from galaxy.util import checkers
from galaxy.webapps.base.controller import BaseUIController
from tool_shed.dependencies import attribute_handlers
from tool_shed.galaxy_install import dependency_display
from tool_shed.metadata import repository_metadata_manager
from tool_shed.repository_types import util as rt_util
from tool_shed.tools.data_table_manager import ShedToolDataTableManager
from tool_shed.util import (
    basic_util,
    commit_util,
    hg_util,
    repository_content_util,
    repository_util,
)
from tool_shed.util import shed_util_common as suc
from tool_shed.util import xml_util
from tool_shed.util.web_util import escape
from tool_shed.webapp.framework.decorators import require_login

log = logging.getLogger(__name__)


[docs]class UploadController(BaseUIController):
[docs] @web.expose @require_login("upload", use_panels=True) def upload(self, trans, **kwd): message = escape(kwd.get("message", "")) status = kwd.get("status", "done") commit_message = escape(kwd.get("commit_message", "Uploaded")) repository_id = kwd.get("repository_id", "") repository = repository_util.get_repository_in_tool_shed(trans.app, repository_id) repo_dir = repository.repo_path(trans.app) uncompress_file = util.string_as_bool(kwd.get("uncompress_file", "true")) remove_repo_files_not_in_tar = util.string_as_bool(kwd.get("remove_repo_files_not_in_tar", "true")) uploaded_file = None upload_point = commit_util.get_upload_point(repository, **kwd) tip = repository.tip() file_data = kwd.get("file_data", "") url = kwd.get("url", "") # Part of the upload process is sending email notification to those that have registered to # receive them. One scenario occurs when the first change set is produced for the repository. # See the suc.handle_email_alerts() method for the definition of the scenarios. new_repo_alert = repository.is_new() uploaded_directory = None if kwd.get("upload_button", False): if file_data == "" and url == "": message = "No files were entered on the upload form." status = "error" uploaded_file = None elif url and url.startswith("hg"): # Use mercurial clone to fetch repository, contents will then be copied over. uploaded_directory = tempfile.mkdtemp() repo_url = f"http{url[len('hg'):]}" cloned_ok, error_message = hg_util.clone_repository(repo_url, uploaded_directory) if not cloned_ok: message = f"Error uploading via mercurial clone: {error_message}" status = "error" basic_util.remove_dir(uploaded_directory) uploaded_directory = None elif url: valid_url = True try: stream = requests.get(url, stream=True, timeout=util.DEFAULT_SOCKET_TIMEOUT) except Exception as e: valid_url = False message = f"Error uploading file via http: {util.unicodify(e)}" status = "error" uploaded_file = None if valid_url: with tempfile.NamedTemporaryFile(mode="wb", delete=False) as uploaded_file: uploaded_file_name = uploaded_file.name for chunk in stream.iter_content(chunk_size=util.CHUNK_SIZE): if chunk: uploaded_file.write(chunk) uploaded_file.flush() uploaded_file_filename = url.split("/")[-1] isempty = os.path.getsize(os.path.abspath(uploaded_file_name)) == 0 elif file_data not in ("", None): uploaded_file = file_data.file uploaded_file_name = uploaded_file.name uploaded_file_filename = os.path.split(file_data.filename)[-1] isempty = os.path.getsize(os.path.abspath(uploaded_file_name)) == 0 if uploaded_file or uploaded_directory: rdah = attribute_handlers.RepositoryDependencyAttributeHandler(trans.app, unpopulate=False) tdah = attribute_handlers.ToolDependencyAttributeHandler(trans.app, unpopulate=False) stdtm = ShedToolDataTableManager(trans.app) ok = True isgzip = False isbz2 = False if uploaded_file: if uncompress_file: isgzip = checkers.is_gzip(uploaded_file_name) if not isgzip: isbz2 = checkers.is_bz2(uploaded_file_name) if isempty: tar = None istar = False else: # Determine what we have - a single file or an archive try: if (isgzip or isbz2) and uncompress_file: # Open for reading with transparent compression. tar = tarfile.open(uploaded_file_name, "r:*") else: tar = tarfile.open(uploaded_file_name) istar = True except tarfile.ReadError: tar = None istar = False else: # Uploaded directory istar = False if istar: ( ok, message, files_to_remove, content_alert_str, undesirable_dirs_removed, undesirable_files_removed, ) = repository_content_util.upload_tar( trans, rdah, tdah, repository, tar, uploaded_file, upload_point, remove_repo_files_not_in_tar, commit_message, new_repo_alert, ) elif uploaded_directory: ( ok, message, files_to_remove, content_alert_str, undesirable_dirs_removed, undesirable_files_removed, ) = self.upload_directory( trans, rdah, tdah, repository, uploaded_directory, upload_point, remove_repo_files_not_in_tar, commit_message, new_repo_alert, ) else: if (isgzip or isbz2) and uncompress_file: uploaded_file_filename = commit_util.uncompress( repository, uploaded_file_name, uploaded_file_filename, isgzip=isgzip, isbz2=isbz2 ) if ( repository.type == rt_util.REPOSITORY_SUITE_DEFINITION and uploaded_file_filename != rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME ): ok = False message = "Repositories of type <b>Repository suite definition</b> can only contain a single file named " message += "<b>repository_dependencies.xml</b>." elif ( repository.type == rt_util.TOOL_DEPENDENCY_DEFINITION and uploaded_file_filename != rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME ): ok = False message = "Repositories of type <b>Tool dependency definition</b> can only contain a single file named " message += "<b>tool_dependencies.xml</b>." if ok: if upload_point is not None: full_path = os.path.abspath(os.path.join(repo_dir, upload_point, uploaded_file_filename)) else: full_path = os.path.abspath(os.path.join(repo_dir, uploaded_file_filename)) # Move some version of the uploaded file to the load_point within the repository hierarchy. if uploaded_file_filename in [rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME]: # Inspect the contents of the file to see if toolshed or changeset_revision attributes # are missing and if so, set them appropriately. altered, root_elem, error_message = rdah.handle_tag_attributes(uploaded_file_name) if error_message: ok = False message = error_message status = "error" elif altered: tmp_filename = xml_util.create_and_write_tmp_file(root_elem) shutil.move(tmp_filename, full_path) else: shutil.move(uploaded_file_name, full_path) elif uploaded_file_filename in [rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME]: # Inspect the contents of the file to see if changeset_revision values are # missing and if so, set them appropriately. altered, root_elem, error_message = tdah.handle_tag_attributes(uploaded_file_name) if error_message: ok = False message = error_message status = "error" if ok: if altered: tmp_filename = xml_util.create_and_write_tmp_file(root_elem) shutil.move(tmp_filename, full_path) else: shutil.move(uploaded_file_name, full_path) else: shutil.move(uploaded_file_name, full_path) if ok: # See if any admin users have chosen to receive email alerts when a repository is updated. # If so, check every uploaded file to ensure content is appropriate. check_contents = commit_util.check_file_contents_for_email_alerts(trans.app) if check_contents and os.path.isfile(full_path): content_alert_str = commit_util.check_file_content_for_html_and_images(full_path) else: content_alert_str = "" hg_util.add_changeset(repo_dir, full_path) hg_util.commit_changeset( repo_dir, full_path_to_changeset=full_path, username=trans.user.username, message=commit_message, ) if full_path.endswith("tool_data_table_conf.xml.sample"): # Handle the special case where a tool_data_table_conf.xml.sample file is being uploaded # by parsing the file and adding new entries to the in-memory trans.app.tool_data_tables # dictionary. error, error_message = stdtm.handle_sample_tool_data_table_conf_file( full_path, persist=False ) if error: message = f"{message}<br/>{error_message}" # See if the content of the change set was valid. admin_only = len(repository.downloadable_revisions) != 1 suc.handle_email_alerts( trans.app, trans.request.host, repository, content_alert_str=content_alert_str, new_repo_alert=new_repo_alert, admin_only=admin_only, ) if ok: # Update the repository files for browsing. hg_util.update_repository(repo_dir) # Get the new repository tip. if tip == repository.tip(): message = "No changes to repository. " status = "warning" else: if (isgzip or isbz2) and uncompress_file: uncompress_str = " uncompressed and " else: uncompress_str = " " if uploaded_directory: source_type = "repository" source = url else: source_type = "file" source = uploaded_file_filename message = f"The {source_type} <b>{escape(source)}</b> has been successfully{uncompress_str}uploaded to the repository. " if istar and (undesirable_dirs_removed or undesirable_files_removed): items_removed = undesirable_dirs_removed + undesirable_files_removed message += ( " %d undesirable items (.hg .svn .git directories, .DS_Store, hgrc files, etc) " % items_removed ) message += "were removed from the archive. " if istar and remove_repo_files_not_in_tar and files_to_remove: if upload_point is not None: message += ( " %d files were removed from the repository relative to the selected upload point '%s'. " % (len(files_to_remove), upload_point) ) else: message += " %d files were removed from the repository root. " % len(files_to_remove) rmm = repository_metadata_manager.RepositoryMetadataManager( app=trans.app, user=trans.user, repository=repository ) status, error_message = rmm.set_repository_metadata_due_to_new_tip( trans.request.host, content_alert_str=content_alert_str, **kwd ) if error_message: message = error_message kwd["message"] = message if repository.metadata_revisions: # A repository's metadata revisions are order descending by update_time, so the zeroth revision # will be the tip just after an upload. metadata_dict = repository.metadata_revisions[0].metadata else: metadata_dict = {} dd = dependency_display.DependencyDisplayer(trans.app) if str(repository.type) not in [ rt_util.REPOSITORY_SUITE_DEFINITION, rt_util.TOOL_DEPENDENCY_DEFINITION, ]: change_repository_type_message = rt_util.generate_message_for_repository_type_change( trans.app, repository ) if change_repository_type_message: message += change_repository_type_message status = "warning" else: # Provide a warning message if a tool_dependencies.xml file is provided, but tool dependencies # weren't loaded due to a requirement tag mismatch or some other problem. Tool dependency # definitions can define orphan tool dependencies (no relationship to any tools contained in the # repository), so warning messages are important because orphans are always valid. The repository # owner must be warned in case they did not intend to define an orphan dependency, but simply # provided incorrect information (tool shed, name owner, changeset_revision) for the definition. orphan_message = dd.generate_message_for_orphan_tool_dependencies(repository, metadata_dict) if orphan_message: message += orphan_message status = "warning" # Handle messaging for invalid tool dependencies. invalid_tool_dependencies_message = dd.generate_message_for_invalid_tool_dependencies(metadata_dict) if invalid_tool_dependencies_message: message += invalid_tool_dependencies_message status = "error" # Handle messaging for invalid repository dependencies. invalid_repository_dependencies_message = dd.generate_message_for_invalid_repository_dependencies( metadata_dict, error_from_tuple=True ) if invalid_repository_dependencies_message: message += invalid_repository_dependencies_message status = "error" # Reset the tool_data_tables by loading the empty tool_data_table_conf.xml file. stdtm.reset_tool_data_tables() if uploaded_directory: basic_util.remove_dir(uploaded_directory) trans.response.send_redirect( web.url_for( controller="repository", action="browse_repository", id=repository_id, commit_message="Deleted selected files", message=message, status=status, ) ) else: if uploaded_directory: basic_util.remove_dir(uploaded_directory) status = "error" # Reset the tool_data_tables by loading the empty tool_data_table_conf.xml file. stdtm.reset_tool_data_tables() return trans.fill_template( "/webapps/tool_shed/repository/upload.mako", repository=repository, changeset_revision=tip, url=url, commit_message=commit_message, uncompress_file=uncompress_file, remove_repo_files_not_in_tar=remove_repo_files_not_in_tar, message=message, status=status, )
[docs] def upload_directory( self, trans, rdah, tdah, repository, uploaded_directory, upload_point, remove_repo_files_not_in_tar, commit_message, new_repo_alert, ): repo_dir = repository.repo_path(trans.app) undesirable_dirs_removed = 0 undesirable_files_removed = 0 if upload_point is not None: full_path = os.path.abspath(os.path.join(repo_dir, upload_point)) else: full_path = os.path.abspath(repo_dir) filenames_in_archive = [] for root, _dirs, files in os.walk(uploaded_directory): for uploaded_file in files: relative_path = os.path.normpath(os.path.join(os.path.relpath(root, uploaded_directory), uploaded_file)) if repository.type == rt_util.REPOSITORY_SUITE_DEFINITION: ok = os.path.basename(uploaded_file) == rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME elif repository.type == rt_util.TOOL_DEPENDENCY_DEFINITION: ok = os.path.basename(uploaded_file) == rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME else: ok = os.path.basename(uploaded_file) not in commit_util.UNDESIRABLE_FILES if ok: for file_path_item in relative_path.split("/"): if file_path_item in commit_util.UNDESIRABLE_DIRS: undesirable_dirs_removed += 1 ok = False break else: undesirable_files_removed += 1 if ok: uploaded_file_name = os.path.abspath(os.path.join(root, uploaded_file)) if os.path.split(uploaded_file_name)[-1] == rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME: # Inspect the contents of the file to see if toolshed or changeset_revision # attributes are missing and if so, set them appropriately. altered, root_elem, error_message = rdah.handle_tag_attributes(uploaded_file_name) if error_message: return False, error_message, [], "", [], [] elif altered: tmp_filename = xml_util.create_and_write_tmp_file(root_elem) shutil.move(tmp_filename, uploaded_file_name) elif os.path.split(uploaded_file_name)[-1] == rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME: # Inspect the contents of the file to see if toolshed or changeset_revision # attributes are missing and if so, set them appropriately. altered, root_elem, error_message = tdah.handle_tag_attributes(uploaded_file_name) if error_message: return False, error_message, [], "", [], [] if altered: tmp_filename = xml_util.create_and_write_tmp_file(root_elem) shutil.move(tmp_filename, uploaded_file_name) repo_path = os.path.join(full_path, relative_path) repo_basedir = os.path.normpath(os.path.join(repo_path, os.path.pardir)) if not os.path.exists(repo_basedir): os.makedirs(repo_basedir) if os.path.exists(repo_path): if os.path.isdir(repo_path): shutil.rmtree(repo_path) else: os.remove(repo_path) shutil.move(os.path.join(uploaded_directory, relative_path), repo_path) filenames_in_archive.append(relative_path) return commit_util.handle_directory_changes( trans.app, trans.request.host, trans.user.username, repository, full_path, filenames_in_archive, remove_repo_files_not_in_tar, new_repo_alert, commit_message, undesirable_dirs_removed, undesirable_files_removed, )