Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/cwltool/resolver.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 """Resolves references to CWL documents from local or remote places.""" | |
2 | |
3 import os | |
4 import urllib | |
5 from pathlib import Path | |
6 from typing import Optional | |
7 | |
8 from schema_salad.ref_resolver import Loader | |
9 | |
10 from .loghandler import _logger | |
11 | |
12 | |
13 def resolve_local(document_loader: Optional[Loader], uri: str) -> Optional[str]: | |
14 pathpart, frag = urllib.parse.urldefrag(uri) | |
15 | |
16 try: | |
17 pathobj = Path(pathpart).resolve() | |
18 except OSError: | |
19 _logger.debug("local resolver could not resolve %s", uri) | |
20 return None | |
21 | |
22 if pathobj.is_file(): | |
23 if frag: | |
24 return f"{pathobj.as_uri()}#{frag}" | |
25 return pathobj.as_uri() | |
26 | |
27 sharepaths = [ | |
28 os.environ.get( | |
29 "XDG_DATA_HOME", os.path.join(os.path.expanduser("~"), ".local", "share") | |
30 ) | |
31 ] | |
32 sharepaths.extend( | |
33 os.environ.get("XDG_DATA_DIRS", "/usr/local/share/:/usr/share/").split(":") | |
34 ) | |
35 shares = [os.path.join(s, "commonwl", uri) for s in sharepaths] | |
36 | |
37 _logger.debug("Search path is %s", shares) | |
38 | |
39 for path in shares: | |
40 if os.path.exists(path): | |
41 return Path(uri).as_uri() | |
42 if os.path.exists(f"{path}.cwl"): | |
43 return Path(f"{path}.cwl").as_uri() | |
44 return None | |
45 | |
46 | |
47 def tool_resolver(document_loader: Loader, uri: str) -> Optional[str]: | |
48 for r in [resolve_local, resolve_ga4gh_tool]: | |
49 ret = r(document_loader, uri) | |
50 if ret is not None: | |
51 return ret | |
52 return None | |
53 | |
54 | |
55 ga4gh_tool_registries = ["https://dockstore.org/api"] | |
56 # in the TRS registry, a primary descriptor can be reached at {0}/api/ga4gh/v2/tools/{1}/versions/{2}/plain-CWL/descriptor | |
57 # The primary descriptor is a CommandLineTool in the case that the files endpoint only describes one file | |
58 # When the primary descriptor is a Workflow, files need to be imported without stripping off "descriptor", looking at the files endpoint is a workaround | |
59 # tested with TRS version 2.0.0-beta.2 | |
60 # TODO not stripping off "descriptor" when looking for local imports would also work https://github.com/ga4gh/tool-registry-service-schemas/blob/2.0.0-beta.2/src/main/resources/swagger/ga4gh-tool-discovery.yaml#L273 | |
61 GA4GH_TRS_FILES = "{0}/api/ga4gh/v2/tools/{1}/versions/{2}/CWL/files" | |
62 GA4GH_TRS_PRIMARY_DESCRIPTOR = ( | |
63 "{0}/api/ga4gh/v2/tools/{1}/versions/{2}/plain-CWL/descriptor/{3}" | |
64 ) | |
65 | |
66 | |
67 def resolve_ga4gh_tool(document_loader: Loader, uri: str) -> Optional[str]: | |
68 path, version = uri.partition(":")[::2] | |
69 if not version: | |
70 version = "latest" | |
71 for reg in ga4gh_tool_registries: | |
72 ds = GA4GH_TRS_FILES.format( | |
73 reg, urllib.parse.quote(path, ""), urllib.parse.quote(version, "") | |
74 ) | |
75 try: | |
76 _logger.debug("Head path is %s", ds) | |
77 resp = document_loader.session.head(ds) | |
78 resp.raise_for_status() | |
79 | |
80 _logger.debug("Passed head path of %s", ds) | |
81 | |
82 resp = document_loader.session.get(ds) | |
83 for file_listing in resp.json(): | |
84 if file_listing.get("file_type") == "PRIMARY_DESCRIPTOR": | |
85 primary_path = file_listing.get("path") | |
86 ds2 = GA4GH_TRS_PRIMARY_DESCRIPTOR.format( | |
87 reg, | |
88 urllib.parse.quote(path, ""), | |
89 urllib.parse.quote(version, ""), | |
90 urllib.parse.quote(primary_path, ""), | |
91 ) | |
92 _logger.debug("Resolved %s", ds2) | |
93 return ds2 | |
94 except Exception: # nosec | |
95 pass | |
96 return None |