Source code for galaxy.managers.datatypes

from typing import (
    Dict,
    List,
    Optional,
    Union,
)

from pydantic.tools import parse_obj_as

from galaxy.datatypes._schema import (
    DatatypeConverterList,
    DatatypeDetails,
    DatatypesCombinedMap,
    DatatypesMap,
)
from galaxy.datatypes.data import Data
from galaxy.datatypes.registry import Registry


[docs]def view_index(
    datatypes_registry: Registry,
    extension_only: Optional[bool] = True,
    upload_only: Optional[bool] = True
) -> Union[List[DatatypeDetails], List[str]]:
    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 upload_only and not datatype_info_dict.get('display_in_upload'):
                continue
            rval.append(datatype_info_dict)
        return rval


[docs]def view_mapping(datatypes_registry: Registry) -> DatatypesMap:
    ext_to_class_name: Dict[str, str] = {}
    classes = []
    for k, v in datatypes_registry.datatypes_by_extension.items():
        c = v.__class__
        ext_to_class_name[k] = f"{c.__module__}.{c.__name__}"
        classes.append(c)
    class_to_classes: Dict[str, Dict[str, bool]] = {}

    def visit_bases(types, cls):
        for base in cls.__bases__:
            if issubclass(base, Data):
                types.add(f"{base.__module__}.{base.__name__}")
            visit_bases(types, base)

    for c in classes:
        n = f"{c.__module__}.{c.__name__}"
        types = {n}
        visit_bases(types, c)
        class_to_classes[n] = {t: True for t in types}
    return DatatypesMap(ext_to_class_name=ext_to_class_name, class_to_classes=class_to_classes)


[docs]def view_types_and_mapping(
    datatypes_registry: Registry,
    extension_only: Optional[bool] = True,
    upload_only: Optional[bool] = True
) -> DatatypesCombinedMap:
    return DatatypesCombinedMap(
        datatypes=view_index(datatypes_registry, extension_only, upload_only),
        datatypes_mapping=view_mapping(datatypes_registry)
    )


[docs]def view_sniffers(datatypes_registry: Registry) -> List[str]:
    rval: List[str] = []
    for sniffer_elem in datatypes_registry.sniffer_elems:
        datatype = sniffer_elem.get('type')
        if datatype is not None:
            rval.append(datatype)
    return rval


[docs]def view_converters(datatypes_registry: Registry) -> DatatypeConverterList:
    converters = []
    for (source_type, targets) in 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 parse_obj_as(DatatypeConverterList, converters)


[docs]def view_edam_formats(datatypes_registry: Registry) -> Dict[str, str]:
    return datatypes_registry.edam_formats


[docs]def view_edam_data(datatypes_registry: Registry) -> Dict[str, str]:
    return datatypes_registry.edam_data