Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/cwltool/resolver.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
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