Mercurial > repos > shellac > sam_consensus_v3
diff 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 |
line wrap: on
line diff
--- /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