This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.workflow.trs_proxy

import logging
import os
import re
import urllib.parse

import requests
import yaml

from galaxy.exceptions import MessageException
from galaxy.util import (
from galaxy.util.search import parse_filters

log = logging.getLogger(__name__)

# Make configurable via YAML,JSON
        "id": "dockstore",
        "api_url": "https://dockstore.org/api",
        "link_url": "https://dockstore.org",
        "label": "Dockstore",
        "doc": "Dockstore is an open platform used by the GA4GH for sharing Docker-based tools and workflows.",

[docs]def parse_search_kwds(search_query): filters = { "organization": "organization", "o": "organization", "name": "name", "n": "name", } keyed_terms, description_term = parse_filters(search_query, filters) query_kwd = { "toolClass": "Workflow", "descriptorType": "GALAXY", } if description_term and description_term.strip(): query_kwd["description"] = (description_term,) if keyed_terms is not None: for key, value, _ in keyed_terms: query_kwd[key] = value return query_kwd
[docs]class TrsProxy:
[docs] def __init__(self, config=None): config_file = getattr(config, "trs_servers_config_file", None) if config_file and os.path.exists(config_file): with open(config_file) as f: server_list = yaml.safe_load(f) else: server_list = DEFAULT_TRS_SERVERS self._server_list = server_list if server_list else [] self._server_dict = {t["id"]: t for t in self._server_list}
[docs] def get_servers(self): return self._server_list
[docs] def get_server(self, trs_server): trs_url = self._server_dict[trs_server]["api_url"] return TrsServer(trs_url)
[docs] def server_from_url(self, trs_url): return TrsServer(trs_url)
[docs] def match_url(self, url): matches = re.match(TRS_URL_REGEX, url) if matches: match_dict = matches.groupdict() match_dict["tool_id"] = urllib.parse.unquote(match_dict["tool_id"]) return match_dict else: return None
[docs]class TrsServer:
[docs] def __init__(self, trs_url): self._trs_url = trs_url
[docs] def get_tools(self, **kwd): query_kwd = {} for key in ["toolClass", "descriptorType", "description", "name", "organization"]: value = kwd.pop(key, None) if value is not None: query_kwd[key] = value trs_api_url = self._get_api_endpoint(**kwd) return self._get(trs_api_url, params=query_kwd)
[docs] def get_tool(self, tool_id, **kwd): trs_api_url = self._get_tool_api_endpoint(tool_id, **kwd) return self._get(trs_api_url)
[docs] def get_versions(self, tool_id, **kwd): trs_api_url = f"{self._get_tool_api_endpoint(tool_id, **kwd)}/versions" return self._get(trs_api_url)
[docs] def get_version(self, tool_id, version_id, **kwd): trs_api_url = f"{self._get_tool_api_endpoint(tool_id, **kwd)}/versions/{version_id}" return self._get(trs_api_url)
[docs] def get_version_descriptor(self, tool_id, version_id, **kwd): trs_api_url = ( f"{self._get_tool_api_endpoint(tool_id, **kwd)}/versions/{version_id}/{GA4GH_GALAXY_DESCRIPTOR}/descriptor" ) return self._get(trs_api_url)["content"]
def _quote(self, tool_id, **kwd): if asbool(kwd.get("tool_id_b64_encoded", False)): import base64 tool_id = base64.b64decode(tool_id) tool_id = urllib.parse.quote_plus(tool_id) return tool_id def _get(self, url, params=None): response = requests.get(url, params=params, timeout=DEFAULT_SOCKET_TIMEOUT) if response.ok: return response.json() else: code = response.status_code message = response.text try: trs_error_dict = response.json() code = int(trs_error_dict["code"]) message = trs_error_dict["message"] except Exception: pass raise MessageException.from_code(code, message) def _get_api_endpoint(self, **kwd): trs_url = self._trs_url trs_api_endpoint = f"{trs_url}/ga4gh/trs/v2/tools" return trs_api_endpoint def _get_tool_api_endpoint(self, tool_id, **kwd): tool_id = self._quote(tool_id, **kwd) trs_api_url = f"{self._get_api_endpoint(**kwd)}/{tool_id}" return trs_api_url