Source code for galaxy.webapps.galaxy.api.datatypes

"""
API operations allowing clients to determine datatype supported by Galaxy.
"""
import logging

from galaxy import exceptions
from galaxy.datatypes.data import Data
from galaxy.util import asbool
from galaxy.web import _future_expose_api_anonymous_and_sessionless as expose_api_anonymous_and_sessionless
from galaxy.web.base.controller import BaseAPIController

log = logging.getLogger(__name__)


[docs]class DatatypesController(BaseAPIController):

[docs]    @expose_api_anonymous_and_sessionless
    def index(self, trans, **kwd):
        """
        GET /api/datatypes
        Return an object containing upload datatypes.
        """
        datatypes_registry = self._datatypes_registry
        try:
            extension_only = asbool(kwd.get('extension_only', True))
            upload_only = asbool(kwd.get('upload_only', True))
            if extension_only:
                if upload_only:
                    return datatypes_registry.upload_file_formats
                else:
                    return [ext for ext in datatypes_registry.datatypes_by_extension]
            else:
                rval = []
                for datatype_info_dict in datatypes_registry.datatype_info_dicts:
                    if not datatype_info_dict.get('display_in_upload') and upload_only:
                        continue
                    rval.append(datatype_info_dict)
                return rval
        except Exception as exception:
            log.error('could not get datatypes: %s', str(exception), exc_info=True)
            if not isinstance(exception, exceptions.MessageException):
                raise exceptions.InternalServerError(str(exception))
            else:
                raise

[docs]    @expose_api_anonymous_and_sessionless
    def mapping(self, trans, **kwd):
        '''
        GET /api/datatypes/mapping
        Return a dictionary of class to class mappings.
        '''
        try:
            ext_to_class_name = dict()
            classes = []
            for k, v in self._datatypes_registry.datatypes_by_extension.items():
                c = v.__class__
                ext_to_class_name[k] = c.__module__ + "." + c.__name__
                classes.append(c)
            class_to_classes = dict()

            def visit_bases(types, cls):
                for base in cls.__bases__:
                    if issubclass(base, Data):
                        types.add(base.__module__ + "." + base.__name__)
                    visit_bases(types, base)
            for c in classes:
                n = c.__module__ + "." + c.__name__
                types = set([n])
                visit_bases(types, c)
                class_to_classes[n] = dict((t, True) for t in types)
            return dict(ext_to_class_name=ext_to_class_name, class_to_classes=class_to_classes)

        except Exception as exception:
            log.error('could not get datatype mapping: %s', str(exception), exc_info=True)
            if not isinstance(exception, exceptions.MessageException):
                raise exceptions.InternalServerError(str(exception))
            else:
                raise

[docs]    @expose_api_anonymous_and_sessionless
    def sniffers(self, trans, **kwd):
        '''
        GET /api/datatypes/sniffers
        Return a list of sniffers.
        '''
        try:
            rval = []
            for sniffer_elem in self._datatypes_registry.sniffer_elems:
                datatype = sniffer_elem.get('type')
                if datatype is not None:
                    rval.append(datatype)
            return rval
        except Exception as exception:
            log.error('could not get datatypes: %s', str(exception), exc_info=True)
            if not isinstance(exception, exceptions.MessageException):
                raise exceptions.InternalServerError(str(exception))
            else:
                raise

[docs]    @expose_api_anonymous_and_sessionless
    def converters(self, trans, **kwd):
        converters = []
        for (source_type, targets) in self._datatypes_registry.datatype_converters.items():
            for target_type in targets:
                converters.append({
                    'source': source_type,
                    'target': target_type,
                    'tool_id': targets[target_type].id,
                })

        return converters

[docs]    @expose_api_anonymous_and_sessionless
    def edam_formats(self, trans, **kwds):
        return self._datatypes_registry.edam_formats

[docs]    @expose_api_anonymous_and_sessionless
    def edam_data(self, trans, **kwds):
        return self._datatypes_registry.edam_data

    @property
    def _datatypes_registry(self):
        return self.app.datatypes_registry