feat: add parser method for TRS URIs

requirements.txt

@@ -2,3 +2,5 @@ coverage==5.2.1
 coveralls==2.1.2
 flake8==3.8.3
 pytest==6.0.1
+requests==2.24.0
+requests-mock==1.8.0

tests/test_client.py

@@ -8,8 +8,11 @@
 
 from trs_cli.client import TRSClient
 from trs_cli.errors import (
-    InvalidURI
+    InvalidURI,
+    InvalidResourcedentifier
 )
+import requests_mock
+import requests
 
 MOCK_HOST = "https://fakehost.com"
 MOCK_TRS_URI = "trs://fakehost.com/SOME_OBJECT"
@@ -24,14 +27,115 @@
     "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.aaaa.com/SOME_OBJECT"
 )
 MOCK_TRS_URI_INVALID = "tr://fakehost.com/SOME_OBJECT"
-MOCK_PORT = 8080
+MOCK_PORT = 443
 MOCK_BASE_PATH = "a/b/c"
+MOCK_ID = "123456"
 MOCK_TOKEN = "MyT0k3n"
-MOCK_TRS_URL = f"{MOCK_HOST}:{MOCK_PORT}/ga4gh/drs/v1/objects"
+MOCK_TRS_URL = f"{MOCK_HOST}:{MOCK_PORT}/a/b/c/tools"
+
+MOCK_TOOL_CLASS = {
+    "description": "description",
+    "id": MOCK_ID,
+    "name": "name",
+}
+
+MOCK_FILES = {
+    "file_wrapper": {
+        "checksum": [
+            {
+                "checksum": "checksum",
+                "type": "sha1",
+            }
+        ],
+        "content": "content",
+        "url": "url",
+    },
+    "tool_file": {
+        "file_type": "PRIMARY_DESCRIPTOR",
+        "path": "path",
+    },
+    "type": "CWL"
+}
+
+MOCK_IMAGES = [
+    {
+        "checksum": [
+            {
+                "checksum": "checksums",
+                "type": "sha256"
+            }
+        ],
+        "image_name": "image_name",
+        "image_type": "Docker",
+        "registry_host": "registry_host",
+        "size": 0,
+        "updated": "updated",
+    }
+]
+
+MOCK_VERSION_NO_ID = {
+    "author": [
+        "author"
+    ],
+    "descriptor_type": [
+        "CWL"
+    ],
+    "files": MOCK_FILES,
+    "images": MOCK_IMAGES,
+    "included_apps": [
+        "https://bio.tools/tool/mytum.de/SNAP2/1",
+        "https://bio.tools/bioexcel_seqqc"
+    ],
+    "is_production": True,
+    "name": "name",
+    "signed": True,
+    "verified_source": [
+        "verified_source",
+    ]
+}
+
+MOCK_TOOL = {
+    "aliases": [
+        "alias_1",
+        "alias_2",
+        "alias_3",
+    ],
+    "checker_url": "checker_url",
+    "description": "description",
+    "has_checker": True,
+    "name": "name",
+    "organization": "organization",
+    "toolclass": MOCK_TOOL_CLASS,
+    "versions": [
+        MOCK_VERSION_NO_ID,
+    ],
+}
+
+MOCK_ERROR = {
+    "msg": "mock_message",
+    "status_code": "400"
+}
+
+MOCK_VERSION_ID = "version"
+
+MOCK_TOOL_ID_AND_VERSION_URL = f"{MOCK_HOST}:{MOCK_PORT}/a/b/c/tools/" \
+    f"{MOCK_ID}/versions/{MOCK_VERSION_ID}"
+
+MOCK_TOOL_ID = f"{MOCK_HOST}:{MOCK_PORT}/a/b/c/tools/{MOCK_ID}"
+
+MOCK_ID_URL = f"trs://fakehost.com/{MOCK_ID}"
+
+MOCK_VERSION_ID_URL = f"trs://fakehost.com/{MOCK_ID}" \
+    f"/versions/{MOCK_VERSION_ID}"
 
 
 class TestTRSClient(unittest.TestCase):
 
+    cli = TRSClient(
+        uri=MOCK_TRS_URI,
+        base_path=MOCK_BASE_PATH,
+    )
+
     def test_cli(self):
         """Test url attribute"""
         cli = TRSClient(
@@ -64,3 +168,79 @@ def test_cli(self):
                 uri=MOCK_TRS_URI_LONG,
                 base_path=MOCK_BASE_PATH,
             )
+
+    def test_get_tool_version(self):
+        """Test get_tool_version url"""
+        with requests_mock.Mocker() as m:
+            m.get(
+                f"{self.cli.uri}/tools/{MOCK_ID}",
+                status_code=200,
+                json=MOCK_TOOL,
+            )
+            self.cli._get_tool_version(tool_id=MOCK_ID)
+            self.assertEqual(
+                m.last_request.url,
+                f"{MOCK_TRS_URL}/{MOCK_ID}",
+            )
+
+            m.get(
+                f"{self.cli.uri}/tools/{MOCK_ID}",
+                status_code=200,
+                json=MOCK_TOOL,
+            )
+            self.cli._get_tool_version(tool_id=MOCK_ID)
+            self.assertEqual(
+                m.last_request.url,
+                f"{MOCK_TRS_URL}/{MOCK_ID}",
+            )
+
+            m.get(
+                f"{self.cli.uri}/tools/{MOCK_ID}",
+                status_code=200,
+                json=MOCK_TOOL,
+            )
+            self.cli._get_tool_version(tool_id=MOCK_ID_URL)
+            self.assertEqual(
+                m.last_request.url,
+                f"{MOCK_TRS_URL}/{MOCK_ID}",
+            )
+
+            m.get(
+                f"{self.cli.uri}/tools/{MOCK_ID}/versions/{MOCK_VERSION_ID}",
+                status_code=200,
+                json=MOCK_TOOL,
+            )
+            self.cli._get_tool_version(tool_id=MOCK_VERSION_ID_URL)
+            self.assertEqual(
+                m.last_request.url,
+                f"{MOCK_TRS_URL}/{MOCK_ID}/versions/{MOCK_VERSION_ID}",
+            )
+
+            m.get(
+                f"{self.cli.uri}/tools/{MOCK_ID}/versions/{MOCK_VERSION_ID}",
+                status_code=200,
+                json=MOCK_TOOL,
+            )
+            self.cli._get_tool_version(
+                tool_id=MOCK_ID, version_id=MOCK_VERSION_ID)
+            self.assertEqual(
+                m.last_request.url,
+                f"{MOCK_TRS_URL}/{MOCK_ID}/versions/{MOCK_VERSION_ID}",
+            )
+
+            m.get(
+                f"{self.cli.uri}/tools/{MOCK_ID}",
+                exc=requests.exceptions.ConnectionError
+            )
+            with pytest.raises(requests.exceptions.ConnectionError):
+                self.cli._get_tool_version(tool_id=MOCK_ID)
+
+            m.get(
+                f"{self.cli.uri}/tools/{MOCK_ID}/versions/{MOCK_VERSION_ID}",
+                status_code=404,
+                text="mock_text",
+            )
+
+            with pytest.raises(InvalidResourcedentifier):
+                self.cli._get_tool_version(
+                    tool_id=MOCK_ID, version_id=MOCK_VERSION_ID)

trs_cli/client.py

@@ -3,11 +3,16 @@
 import logging
 import re
 import sys
+import requests
+import socket
 from typing import (Dict, Optional, Tuple)
+from urllib.parse import quote
+import urllib3
 
 from trs_cli.errors import (
     exception_handler,
     InvalidURI,
+    InvalidResourcedentifier,
 )
 
 logger = logging.getLogger(__name__)
@@ -48,9 +53,12 @@ class TRSClient():
     """
     # set regular expressions as private class variables
     _RE_DOMAIN_PART = r'[a-z0-9]([a-z0-9-]{1,61}[a-z0-9]?)?'
-    _RE_DOMAIN = rf"({_RE_DOMAIN_PART}\.)+{_RE_DOMAIN_PART}\.?"
+    _RE_DOMAIN = rf"({_RE_DOMAIN_PART}\.?)+{_RE_DOMAIN_PART}\.?"
     _RE_TRS_ID = r'\S+'  # TODO: update to account for versioned TRS URIs
     _RE_HOST = rf"^(?P<schema>trs|http|https):\/\/(?P<host>{_RE_DOMAIN})\/?"
+    _RE_TRS_TOOL_UID = r"[a-z0-9]{6}"
+    _RE_TOOL_ID = rf"^(trs:\/\/{_RE_DOMAIN}\/)?(?P<tool_id>" \
+        rf"{_RE_TRS_TOOL_UID})(\/versions\/)?(?P<tool_version_id>.*)?"
 
     def __init__(
         self,
@@ -78,6 +86,81 @@ def __init__(
     #
     # Check DRS-cli repo for examples
 
+    def _get_tool_version(
+        self,
+        tool_id: Optional[str] = None,
+        version_id: Optional[str] = None,
+    ) -> Tuple[str, Optional[str]]:
+        """
+        Return sanitized tool and/or version identifiers or extract them from
+        a TRS URI.
+
+        Arguments:
+            tool_id: Implementation-specific TRS tool identifier OR TRS URI
+                pointing to a given tool, cf.
+                https://ga4gh.github.io/tool-registry-service-schemas/DataModel/#trs_uris
+                Note that if a TRS URI is passed, only the TRS identifier parts
+                (tool and version) will be evaluated. To reset the hostname,
+                create a new client with the `TRSClient()` constructor.
+            version_id: Implementation-specific TRS version identifier; if
+                provided, will take precedence over any version identifier
+                extracted from a versioned TRS URI passed to `tool_id`
+
+            Returns:
+                Tuple of validated, percent-encoded tool and version
+                identifiers, respectively; if no `version_id` was supplied OR
+                an unversioned TRS URI was passed to `tool_id`, the second
+                item of the tuple will be set to `None`; likewise, if not
+                `tool_id` was provided, the first item of the tuple will
+                be set to `None`.
+
+            Raises:
+                drs_cli.errors.InvalidResourcedentifier: input tool identifier
+                    cannot be parsed.
+        """
+        if not tool_id and not version_id:
+            raise InvalidResourcedentifier(
+                "Tool_id and version_id not provided"
+            )
+
+        re_tool_id_regex, re_version_id = self._get_tool_and_version_id(
+            tool_id=tool_id)
+
+        if tool_id and not version_id:
+            if re_version_id:
+                version_id = re_version_id
+                url = f"{self.uri}/tools/{re_tool_id_regex}" \
+                    f"/versions/{version_id}"
+            else:
+                url = f"{self.uri}/tools/{re_tool_id_regex}"
+        elif tool_id and version_id:
+            url = f"{self.uri}/tools/{re_tool_id_regex}/versions/{version_id}"
+
+        logger.info(f"Request URL: {url}")
+
+        try:
+            response = requests.get(
+                url=url,
+                headers=self.headers,
+            )
+        except (
+            requests.exceptions.ConnectionError,
+            socket.gaierror,
+            urllib3.exceptions.NewConnectionError,
+        ):
+            raise requests.exceptions.ConnectionError(
+                "Could not connect to API endpoint."
+            )
+        if not response.status_code == 200:
+            raise InvalidResourcedentifier(
+                "Input tool identifier cannot be parsed"
+            )
+
+        if not version_id:  # If version_id is '' i.e NA according to regex
+            version_id = None
+
+        return [re_tool_id_regex, version_id]
+
     def _get_headers(self) -> Dict:
         """Build dictionary of request headers.
 
@@ -125,3 +208,18 @@ def _get_host(
             return (schema, host)
         else:
             raise InvalidURI
+
+    def _get_tool_and_version_id(
+        self,
+        tool_id: str,
+    ) -> str:
+        """
+        Arguments:
+            tool_id: Implementation-specific TRS identifier OR hostname-based
+               TRS URI pointing to a given object.
+        Returns:
+            Validated, percent-encoded tool id and version id.
+        """
+        match = re.search(self._RE_TOOL_ID, tool_id, re.I)
+        return quote(string=match.group('tool_id'), safe=''), \
+            quote(string=match.group('tool_version_id'), safe='')

trs_cli/errors.py

@@ -19,3 +19,7 @@ def exception_handler(
 
 class InvalidURI(Exception):
     """Exception raised for invalid URIs."""
+
+
+class InvalidResourcedentifier(Exception):
+    """Exception raised when an invalid API response is encountered."""