diff env/lib/python3.9/site-packages/cwltool/resolver.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.9/site-packages/cwltool/resolver.py	Mon Mar 22 18:12:50 2021 +0000
@@ -0,0 +1,96 @@
+"""Resolves references to CWL documents from local or remote places."""
+
+import os
+import urllib
+from pathlib import Path
+from typing import Optional
+
+from schema_salad.ref_resolver import Loader
+
+from .loghandler import _logger
+
+
+def resolve_local(document_loader: Optional[Loader], uri: str) -> Optional[str]:
+    pathpart, frag = urllib.parse.urldefrag(uri)
+
+    try:
+        pathobj = Path(pathpart).resolve()
+    except OSError:
+        _logger.debug("local resolver could not resolve %s", uri)
+        return None
+
+    if pathobj.is_file():
+        if frag:
+            return f"{pathobj.as_uri()}#{frag}"
+        return pathobj.as_uri()
+
+    sharepaths = [
+        os.environ.get(
+            "XDG_DATA_HOME", os.path.join(os.path.expanduser("~"), ".local", "share")
+        )
+    ]
+    sharepaths.extend(
+        os.environ.get("XDG_DATA_DIRS", "/usr/local/share/:/usr/share/").split(":")
+    )
+    shares = [os.path.join(s, "commonwl", uri) for s in sharepaths]
+
+    _logger.debug("Search path is %s", shares)
+
+    for path in shares:
+        if os.path.exists(path):
+            return Path(uri).as_uri()
+        if os.path.exists(f"{path}.cwl"):
+            return Path(f"{path}.cwl").as_uri()
+    return None
+
+
+def tool_resolver(document_loader: Loader, uri: str) -> Optional[str]:
+    for r in [resolve_local, resolve_ga4gh_tool]:
+        ret = r(document_loader, uri)
+        if ret is not None:
+            return ret
+    return None
+
+
+ga4gh_tool_registries = ["https://dockstore.org/api"]
+# in the TRS registry, a primary descriptor can be reached at {0}/api/ga4gh/v2/tools/{1}/versions/{2}/plain-CWL/descriptor
+# The primary descriptor is a CommandLineTool in the case that the files endpoint only describes one file
+# When the primary descriptor is a Workflow, files need to be imported without stripping off "descriptor", looking at the files endpoint is a workaround
+# tested with TRS version 2.0.0-beta.2
+# 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
+GA4GH_TRS_FILES = "{0}/api/ga4gh/v2/tools/{1}/versions/{2}/CWL/files"
+GA4GH_TRS_PRIMARY_DESCRIPTOR = (
+    "{0}/api/ga4gh/v2/tools/{1}/versions/{2}/plain-CWL/descriptor/{3}"
+)
+
+
+def resolve_ga4gh_tool(document_loader: Loader, uri: str) -> Optional[str]:
+    path, version = uri.partition(":")[::2]
+    if not version:
+        version = "latest"
+    for reg in ga4gh_tool_registries:
+        ds = GA4GH_TRS_FILES.format(
+            reg, urllib.parse.quote(path, ""), urllib.parse.quote(version, "")
+        )
+        try:
+            _logger.debug("Head path is %s", ds)
+            resp = document_loader.session.head(ds)
+            resp.raise_for_status()
+
+            _logger.debug("Passed head path of %s", ds)
+
+            resp = document_loader.session.get(ds)
+            for file_listing in resp.json():
+                if file_listing.get("file_type") == "PRIMARY_DESCRIPTOR":
+                    primary_path = file_listing.get("path")
+                    ds2 = GA4GH_TRS_PRIMARY_DESCRIPTOR.format(
+                        reg,
+                        urllib.parse.quote(path, ""),
+                        urllib.parse.quote(version, ""),
+                        urllib.parse.quote(primary_path, ""),
+                    )
+                    _logger.debug("Resolved %s", ds2)
+                    return ds2
+        except Exception:  # nosec
+            pass
+    return None