Mercurial > repos > guerler > springsuite

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

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:18:57 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:d30785e31577
1 """Abstraction around cwltool and related libraries for loading a CWL artifact."""
2 import os
3 from collections import namedtuple
4
5 from .cwltool_deps import (
6 default_loader,
7 ensure_cwltool_available,
8 load_tool,
9 LoadingContext,
10 resolve_and_validate_document,
11 )
12
13 RawProcessReference = namedtuple("RawProcessReference", ["loading_context", "process_object", "uri"])
14 ResolvedProcessDefinition = namedtuple("ResolvedProcessDefinition", ["loading_context", "uri", "raw_process_reference"])
15 REWRITE_EXPRESSIONS = False
16
17
18 class SchemaLoader(object):
19
20 def __init__(self, strict=True, validate=True):
21 self._strict = strict
22 self._validate = validate
23
24 @property
25 def raw_document_loader(self):
26 ensure_cwltool_available()
27 return default_loader(None)
28
29 def loading_context(self):
30 loading_context = LoadingContext()
31 loading_context.strict = self._strict
32 loading_context.do_validate = self._validate
33 loading_context.loader = self.raw_document_loader
34 loading_context.do_update = True
35 return loading_context
36
37 def raw_process_reference(self, path, loading_context=None):
38 path = os.path.abspath(path)
39 uri = "file://" + path
40 loading_context = loading_context or self.loading_context()
41 if REWRITE_EXPRESSIONS and not uri.endswith(".galaxy"):
42 galaxy_path = os.path.abspath(path) + ".galaxy"
43 from cwl_utils import etools_to_clt
44 etools_to_clt.main([path, galaxy_path])
45 galaxy_uri = "file://" + galaxy_path
46 uri = galaxy_uri
47 loading_context, process_object, uri = load_tool.fetch_document(uri, loadingContext=loading_context)
48 return RawProcessReference(loading_context, process_object, uri)
49
50 def raw_process_reference_for_object(self, process_object, uri=None, loading_context=None):
51 if uri is None:
52 uri = "galaxy://"
53 loading_context = loading_context or self.loading_context()
54 process_object["id"] = uri
55 loading_context, process_object, uri = load_tool.fetch_document(process_object, loadingContext=loading_context)
56 return RawProcessReference(loading_context, process_object, uri)
57
58 def process_definition(self, raw_process_reference):
59 assert raw_process_reference.loading_context is not None, "No loading context found for raw_process_reference"
60 loading_context, uri = resolve_and_validate_document(
61 raw_process_reference.loading_context,
62 raw_process_reference.process_object,
63 raw_process_reference.uri,
64 )
65 process_def = ResolvedProcessDefinition(
66 loading_context,
67 uri,
68 raw_process_reference,
69 )
70 return process_def
71
72 def tool(self, **kwds):
73 process_definition = kwds.get("process_definition", None)
74 if process_definition is None:
75 raw_process_reference = kwds.get("raw_process_reference", None)
76 if raw_process_reference is None:
77 raw_process_reference = self.raw_process_reference(kwds["path"])
78 process_definition = self.process_definition(raw_process_reference)
79
80 tool = load_tool.make_tool(
81 process_definition.uri,
82 process_definition.loading_context,
83 )
84 return tool
85
86
87 schema_loader = SchemaLoader()
88 non_strict_non_validating_schema_loader = SchemaLoader(strict=False, validate=False)