Mercurial > repos > guerler > springsuite
comparison 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 |
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) |