Source code for galaxy.tool_util.cwl.schema

"""Abstraction around cwltool and related libraries for loading a CWL artifact."""
import os
import tempfile
from collections import namedtuple

from .cwltool_deps import (
    _has_relax_path_checks_flag,
    default_loader,
    ensure_cwltool_available,
    load_tool,
    LoadingContext,
    resolve_and_validate_document,
)

RawProcessReference = namedtuple("RawProcessReference", ["loading_context", "process_object", "uri"])
ResolvedProcessDefinition = namedtuple("ResolvedProcessDefinition", ["loading_context", "uri", "raw_process_reference"])
REWRITE_EXPRESSIONS = False


[docs]class SchemaLoader:

[docs]    def __init__(self, strict=True, validate=True):
        self._strict = strict
        self._validate = validate

    @property
    def raw_document_loader(self):
        ensure_cwltool_available()
        return default_loader(None)

[docs]    def loading_context(self):
        loading_context = LoadingContext()
        loading_context.strict = self._strict
        loading_context.do_validate = self._validate
        loading_context.loader = self.raw_document_loader
        loading_context.do_update = True
        if _has_relax_path_checks_flag():
            loading_context.relax_path_checks = True
        return loading_context

[docs]    def raw_process_reference(self, path, loading_context=None):
        with tempfile.TemporaryDirectory() as output_dir:
            suffix = ''
            if '#' in path:
                path, suffix = path.split('#')
            print(f""" -------

{open(path).read()}

            -------
            """)
            processed_path = os.path.join(output_dir, os.path.basename(path))
            path = os.path.abspath(path)
            uri = "file://" + path
            loading_context = loading_context or self.loading_context()
            if REWRITE_EXPRESSIONS:
                from cwl_utils import cwl_v1_0_expression_refactor
                exit_code = cwl_v1_0_expression_refactor.main([output_dir, path, '--skip-some1', '--skip-some2'])
                if exit_code == 0:
                    uri = f"file://{processed_path}"
                    print(f""" -------

{open(processed_path).read()}

            -------
                    """)
            if suffix:
                uri = f"{uri}#{suffix}"
            loading_context, process_object, uri = load_tool.fetch_document(uri, loadingContext=loading_context)
        return RawProcessReference(loading_context, process_object, uri)

[docs]    def raw_process_reference_for_object(self, process_object, uri=None, loading_context=None):
        if uri is None:
            uri = "galaxy://"
        loading_context = loading_context or self.loading_context()
        process_object["id"] = uri
        loading_context, process_object, uri = load_tool.fetch_document(process_object, loadingContext=loading_context)
        return RawProcessReference(loading_context, process_object, uri)

[docs]    def process_definition(self, raw_process_reference):
        assert raw_process_reference.loading_context is not None, "No loading context found for raw_process_reference"
        loading_context, uri = resolve_and_validate_document(
            raw_process_reference.loading_context,
            raw_process_reference.process_object,
            raw_process_reference.uri,
        )
        process_def = ResolvedProcessDefinition(
            loading_context,
            uri,
            raw_process_reference,
        )
        return process_def

[docs]    def tool(self, **kwds):
        process_definition = kwds.get("process_definition", None)
        if process_definition is None:
            raw_process_reference = kwds.get("raw_process_reference", None)
            if raw_process_reference is None:
                raw_process_reference = self.raw_process_reference(kwds["path"])
            process_definition = self.process_definition(raw_process_reference)

        tool = load_tool.make_tool(
            process_definition.uri,
            process_definition.loading_context,
        )
        return tool


schema_loader = SchemaLoader()
non_strict_non_validating_schema_loader = SchemaLoader(strict=False, validate=False)