Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/galaxy/tool_util/cwl/schema.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 """Abstraction around cwltool and related libraries for loading a CWL artifact.""" | |
2 import os | |
3 import tempfile | |
4 from collections import namedtuple | |
5 | |
6 from .cwltool_deps import ( | |
7 _has_relax_path_checks_flag, | |
8 default_loader, | |
9 ensure_cwltool_available, | |
10 load_tool, | |
11 LoadingContext, | |
12 resolve_and_validate_document, | |
13 ) | |
14 | |
15 RawProcessReference = namedtuple("RawProcessReference", ["loading_context", "process_object", "uri"]) | |
16 ResolvedProcessDefinition = namedtuple("ResolvedProcessDefinition", ["loading_context", "uri", "raw_process_reference"]) | |
17 REWRITE_EXPRESSIONS = False | |
18 | |
19 | |
20 class SchemaLoader: | |
21 | |
22 def __init__(self, strict=True, validate=True): | |
23 self._strict = strict | |
24 self._validate = validate | |
25 | |
26 @property | |
27 def raw_document_loader(self): | |
28 ensure_cwltool_available() | |
29 return default_loader(None) | |
30 | |
31 def loading_context(self): | |
32 loading_context = LoadingContext() | |
33 loading_context.strict = self._strict | |
34 loading_context.do_validate = self._validate | |
35 loading_context.loader = self.raw_document_loader | |
36 loading_context.do_update = True | |
37 if _has_relax_path_checks_flag(): | |
38 loading_context.relax_path_checks = True | |
39 return loading_context | |
40 | |
41 def raw_process_reference(self, path, loading_context=None): | |
42 with tempfile.TemporaryDirectory() as output_dir: | |
43 suffix = '' | |
44 if '#' in path: | |
45 path, suffix = path.split('#') | |
46 print(f""" ------- | |
47 | |
48 {open(path).read()} | |
49 | |
50 ------- | |
51 """) | |
52 processed_path = os.path.join(output_dir, os.path.basename(path)) | |
53 path = os.path.abspath(path) | |
54 uri = "file://" + path | |
55 loading_context = loading_context or self.loading_context() | |
56 if REWRITE_EXPRESSIONS: | |
57 from cwl_utils import cwl_v1_0_expression_refactor | |
58 exit_code = cwl_v1_0_expression_refactor.main([output_dir, path, '--skip-some1', '--skip-some2']) | |
59 if exit_code == 0: | |
60 uri = f"file://{processed_path}" | |
61 print(f""" ------- | |
62 | |
63 {open(processed_path).read()} | |
64 | |
65 ------- | |
66 """) | |
67 if suffix: | |
68 uri = f"{uri}#{suffix}" | |
69 loading_context, process_object, uri = load_tool.fetch_document(uri, loadingContext=loading_context) | |
70 return RawProcessReference(loading_context, process_object, uri) | |
71 | |
72 def raw_process_reference_for_object(self, process_object, uri=None, loading_context=None): | |
73 if uri is None: | |
74 uri = "galaxy://" | |
75 loading_context = loading_context or self.loading_context() | |
76 process_object["id"] = uri | |
77 loading_context, process_object, uri = load_tool.fetch_document(process_object, loadingContext=loading_context) | |
78 return RawProcessReference(loading_context, process_object, uri) | |
79 | |
80 def process_definition(self, raw_process_reference): | |
81 assert raw_process_reference.loading_context is not None, "No loading context found for raw_process_reference" | |
82 loading_context, uri = resolve_and_validate_document( | |
83 raw_process_reference.loading_context, | |
84 raw_process_reference.process_object, | |
85 raw_process_reference.uri, | |
86 ) | |
87 process_def = ResolvedProcessDefinition( | |
88 loading_context, | |
89 uri, | |
90 raw_process_reference, | |
91 ) | |
92 return process_def | |
93 | |
94 def tool(self, **kwds): | |
95 process_definition = kwds.get("process_definition", None) | |
96 if process_definition is None: | |
97 raw_process_reference = kwds.get("raw_process_reference", None) | |
98 if raw_process_reference is None: | |
99 raw_process_reference = self.raw_process_reference(kwds["path"]) | |
100 process_definition = self.process_definition(raw_process_reference) | |
101 | |
102 tool = load_tool.make_tool( | |
103 process_definition.uri, | |
104 process_definition.loading_context, | |
105 ) | |
106 return tool | |
107 | |
108 | |
109 schema_loader = SchemaLoader() | |
110 non_strict_non_validating_schema_loader = SchemaLoader(strict=False, validate=False) |