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,
    unicodify
)
from galaxy.web import 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 e: log.exception('Could not get datatypes') if not isinstance(e, exceptions.MessageException): raise exceptions.InternalServerError(unicodify(e)) 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 e: log.exception('Could not get datatype mapping') if not isinstance(e, exceptions.MessageException): raise exceptions.InternalServerError(unicodify(e)) 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 e: log.exception('Could not get datatypes') if not isinstance(e, exceptions.MessageException): raise exceptions.InternalServerError(unicodify(e)) 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