Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/cwltool/resolver.py @ 5:9b1c78e6ba9c draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author shellac
date Mon, 01 Jun 2020 08:59:25 -0400
parents 79f47841a781
children
comparison
equal deleted inserted replaced
4:79f47841a781 5:9b1c78e6ba9c
1 """Resolves references to CWL documents from local or remote places."""
2 from __future__ import absolute_import
3
4 import os
5 import sys
6
7 from typing import Any, Optional, Text
8
9 from six.moves import urllib
10
11 from schema_salad.ref_resolver import Loader
12
13 from .loghandler import _logger
14 from schema_salad.ref_resolver import Loader
15
16 from pathlib2 import Path
17
18 if not getattr(__builtins__, "WindowsError", None):
19 class WindowsError(OSError): pass
20
21 def resolve_local(document_loader, uri):
22 # type: (Loader, Text) -> Optional[Text]
23 pathpart, frag = urllib.parse.urldefrag(uri)
24
25 try:
26 pathobj = Path(pathpart).resolve()
27 except (WindowsError, OSError):
28 _logger.debug("local resolver could not resolve %s", uri)
29 return None
30
31 if pathobj.is_file():
32 if frag:
33 return "{}#{}".format(pathobj.as_uri(), frag)
34 return pathobj.as_uri()
35
36 sharepaths = [os.environ.get("XDG_DATA_HOME", os.path.join(
37 os.path.expanduser('~'), ".local", "share"))]
38 sharepaths.extend(os.environ.get(
39 "XDG_DATA_DIRS", "/usr/local/share/:/usr/share/").split(":"))
40 shares = [os.path.join(s, "commonwl", uri) for s in sharepaths]
41
42 _logger.debug("Search path is %s", shares)
43
44 for path in shares:
45 if os.path.exists(path):
46 return Path(uri).as_uri()
47 if os.path.exists("{}.cwl".format(path)):
48 return Path("{}.cwl".format(path)).as_uri()
49 return None
50
51
52 def tool_resolver(document_loader, uri): # type: (Loader, Text) -> Optional[Text]
53 for r in [resolve_local, resolve_ga4gh_tool]:
54 ret = r(document_loader, uri)
55 if ret is not None:
56 return ret
57 return None
58
59
60 ga4gh_tool_registries = ["https://dockstore.org/api"]
61 # in the TRS registry, a primary descriptor can be reached at {0}/api/ga4gh/v2/tools/{1}/versions/{2}/plain-CWL/descriptor
62 # The primary descriptor is a CommandLineTool in the case that the files endpoint only describes one file
63 # When the primary descriptor is a Workflow, files need to be imported without stripping off "descriptor", looking at the files endpoint is a workaround
64 # tested with TRS version 2.0.0-beta.2
65 # 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
66 GA4GH_TRS_FILES = "{0}/api/ga4gh/v2/tools/{1}/versions/{2}/CWL/files"
67 GA4GH_TRS_PRIMARY_DESCRIPTOR = "{0}/api/ga4gh/v2/tools/{1}/versions/{2}/plain-CWL/descriptor/{3}"
68
69
70 def resolve_ga4gh_tool(document_loader, uri): # type: (Loader, Text) -> Optional[Text]
71 path, version = uri.partition(":")[::2]
72 if not version:
73 version = "latest"
74 for reg in ga4gh_tool_registries:
75 ds = GA4GH_TRS_FILES.format(reg, urllib.parse.quote(path, ""), urllib.parse.quote(version, ""))
76 try:
77 _logger.debug("Head path is %s", ds)
78 resp = document_loader.session.head(ds)
79 resp.raise_for_status()
80
81 _logger.debug("Passed head path of %s", ds)
82
83 resp = document_loader.session.get(ds)
84 for file_listing in resp.json():
85 if file_listing.get('file_type') == 'PRIMARY_DESCRIPTOR':
86 primary_path = file_listing.get('path')
87 ds2 = GA4GH_TRS_PRIMARY_DESCRIPTOR.format(reg, urllib.parse.quote(path, ""), urllib.parse.quote(version, ""), urllib.parse.quote(primary_path, ""))
88 _logger.debug("Resolved %s", ds2)
89 return ds2
90 except Exception: # nosec
91 pass
92 return None