Source code for galaxy.datatypes.upload_util

import os

from galaxy.datatypes import sniff
from galaxy.util.checkers import (
    check_binary,
    is_single_file_zip,
    is_zip,
)


[docs]class UploadProblemException(Exception):
    pass


[docs]def handle_upload(
    registry,
    path,  # dataset.path
    requested_ext,  # dataset.file_type
    name,  # dataset.name,
    tmp_prefix,
    tmp_dir,
    check_content,
    link_data_only,
    in_place,
    auto_decompress,
    convert_to_posix_lines,
    convert_spaces_to_tabs,
):
    stdout = None
    converted_path = None
    multi_file_zip = False

    # Does the first 1K contain a null?
    is_binary = check_binary(path)

    # Decompress if needed/desired and determine/validate filetype. If a keep-compressed datatype is explicitly selected
    # or if autodetection is selected and the file sniffs as a keep-compressed datatype, it will not be decompressed.
    if not link_data_only:
        if auto_decompress and is_zip(path) and not is_single_file_zip(path):
            multi_file_zip = True
        try:
            ext, converted_path, compression_type = sniff.handle_uploaded_dataset_file_internal(
                path,
                registry,
                ext=requested_ext,
                tmp_prefix=tmp_prefix,
                tmp_dir=tmp_dir,
                in_place=in_place,
                check_content=check_content,
                is_binary=is_binary,
                auto_decompress=auto_decompress,
                uploaded_file_ext=os.path.splitext(name)[1].lower().lstrip('.'),
                convert_to_posix_lines=convert_to_posix_lines,
                convert_spaces_to_tabs=convert_spaces_to_tabs,
            )
        except sniff.InappropriateDatasetContentError as exc:
            raise UploadProblemException(exc)
    elif requested_ext == 'auto':
        ext = sniff.guess_ext(path, registry.sniff_order, is_binary=is_binary)
    else:
        ext = requested_ext

    # The converted path will be the same as the input path if no conversion was done (or in-place conversion is used)
    converted_path = None if converted_path == path else converted_path

    # Validate datasets where the filetype was explicitly set using the filetype's sniffer (if any)
    if requested_ext != 'auto':
        datatype = registry.get_datatype_by_extension(requested_ext)
        # Enable sniffer "validate mode" (prevents certain sniffers from disabling themselves)
        if check_content and hasattr(datatype, 'sniff') and not datatype.sniff(path):
            stdout = ("Warning: The file 'Type' was set to '{ext}' but the file does not appear to be of that"
                      " type".format(ext=requested_ext))

    # Handle unsniffable binaries
    if is_binary and ext == 'binary':
        upload_ext = os.path.splitext(name)[1].lower().lstrip('.')
        if registry.is_extension_unsniffable_binary(upload_ext):
            stdout = ("Warning: The file's datatype cannot be determined from its contents and was guessed based on"
                     " its extension, to avoid this warning, manually set the file 'Type' to '{ext}' when uploading"
                     " this type of file".format(ext=upload_ext))
            ext = upload_ext
        else:
            stdout = ("The uploaded binary file format cannot be determined automatically, please set the file 'Type'"
                      " manually")

    datatype = registry.get_datatype_by_extension(ext)
    if multi_file_zip and not getattr(datatype, 'compressed', False):
        stdout = 'ZIP file contained more than one file, only the first file was added to Galaxy.'

    return stdout, ext, datatype, is_binary, converted_path