diff planemo/lib/python3.7/site-packages/galaxy/tool_util/cwl/schema.py @ 0:d30785e31577 draft

"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:18:57 -0400
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/planemo/lib/python3.7/site-packages/galaxy/tool_util/cwl/schema.py	Fri Jul 31 00:18:57 2020 -0400
@@ -0,0 +1,88 @@
+"""Abstraction around cwltool and related libraries for loading a CWL artifact."""
+import os
+from collections import namedtuple
+
+from .cwltool_deps import (
+    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
+
+
+class SchemaLoader(object):
+
+    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)
+
+    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
+        return loading_context
+
+    def raw_process_reference(self, path, loading_context=None):
+        path = os.path.abspath(path)
+        uri = "file://" + path
+        loading_context = loading_context or self.loading_context()
+        if REWRITE_EXPRESSIONS and not uri.endswith(".galaxy"):
+            galaxy_path = os.path.abspath(path) + ".galaxy"
+            from cwl_utils import etools_to_clt
+            etools_to_clt.main([path, galaxy_path])
+            galaxy_uri = "file://" + galaxy_path
+            uri = galaxy_uri
+        loading_context, process_object, uri = load_tool.fetch_document(uri, loadingContext=loading_context)
+        return RawProcessReference(loading_context, process_object, uri)
+
+    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)
+
+    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
+
+    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)