feat: implement variant match search (issue #114)

pyproject.toml

@@ -147,7 +147,8 @@ ignore-variadic-names = true
 [tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["F401", "E402"]
 "tests/*" = ["ANN001", "ANN2", "ANN102", "S101", "B011", "INP001", "D", "C400", "PLR2004", "PLC0415", "S106", "S107", "SLF001"]
-"src/anyvlm/utils/exceptions.py" = ["S105"] # allows using the word 'token'
+"src/anyvlm/utils/exceptions.py" = ["S105"] # allows using the word 'token',
+"src/anyvlm/scripts/run_variant_match.py" = ["T201", "T203"] # allow printing
 
 [tool.ruff.format]
 docstring-code-format = true

src/anyvlm/outbound/node_registry.py

@@ -13,7 +13,7 @@
 
 
 @dataclass
-class VlmNode:
+class VLMNode:
     """Houses metadata about a VLM Network node"""
 
     client_id: str
@@ -29,7 +29,7 @@ class NodeRegistry:
         100  # Auth0's node list is paginated w/ a max of 100 results per page
     )
 
-    _nodes: list[VlmNode]
+    _nodes: list[VLMNode]
     _nodes_expiry: datetime
     _vlm_client: VLMClient
 
@@ -41,7 +41,7 @@ def __init__(self, vlm_client: VLMClient) -> None:
         self._vlm_client = vlm_client
         self.refresh_nodes()
 
-    def get_nodes(self) -> list[VlmNode]:
+    def get_nodes(self) -> list[VLMNode]:
         """Get a list of known nodes on the VLM network. Uses the cached list if cache is set & unexpired.
         See 'self.refresh_nodes()' for list of possible Exceptions that may be raised.
         """
@@ -60,7 +60,7 @@ def refresh_nodes(self) -> None:
         :raises: `NodeRetrievalError` if request to the VLM Network to obtain the list of registered nodes fails
         :raises: `TokenRetrievalError` if unable to obtain a JWT token to send the request to the VLM Network
         """
-        refreshed_nodes: list[VlmNode] = []
+        refreshed_nodes: list[VLMNode] = []
         page = 0
         while True:
             response: Response = self._vlm_client.request(
@@ -77,7 +77,7 @@ def refresh_nodes(self) -> None:
 
             data = response.json()
             refreshed_nodes.extend(
-                VlmNode(
+                VLMNode(
                     client_id=entry.get("client_id", ""),
                     client_name=entry.get("name", ""),
                     match_url=entry.get("client_metadata", {}).get("match_url"),

src/anyvlm/outbound/variant_match_search.py

@@ -0,0 +1,36 @@
+"""Search for a matching variant in other VLM Nodes, and retrieve their CAF data for this variant"""
+
+from http import HTTPMethod
+
+from requests.models import Response
+
+from anyvlm.auth.token_provider import TokenProvider
+from anyvlm.outbound.vlm_client import VLMClient
+from anyvlm.utils.types import (
+    ChromosomeName,
+    GrcAssemblyId,
+    Nucleotide,
+)
+
+
+def variant_match_search(
+    query_url: str,
+    assembly_id: GrcAssemblyId,
+    reference_name: ChromosomeName,
+    start: int,
+    reference_bases: Nucleotide,
+    alternate_bases: Nucleotide,
+) -> Response:
+    """Search for a matching variant in other VLM Nodes, and retrieve their CAF data for this variant.
+
+    :param query_url: The URL for the node you are querying
+    :assembly_id: The assembly for variant you're matching
+    :reference_name: The name of the chromosome the variant you're matching is on
+    :start: The start position of the variant you're matching
+    :reference_bases: The reference base for the variant you're matching
+    :alternate_bases: The alternate base for hte variant you're matching
+    :return: The Response object from the node you're querying. Will match
+    """
+    query_url += f"?assemblyId={assembly_id}&referenceName={reference_name}&start={start}&referenceBases={reference_bases}&alternateBases={alternate_bases}"
+    vlm_client: VLMClient = VLMClient(token_provider=TokenProvider())
+    return vlm_client.request(method_type=HTTPMethod.GET, url=query_url)

src/anyvlm/scripts/__init__.py

@@ -0,0 +1 @@
+"""Utility scripts"""

src/anyvlm/scripts/run_variant_match.py

@@ -0,0 +1,267 @@
+"""Prompt a user to query an external VLM Network node
+Run this script with the following command: `python -m src.anyvlm.scripts.run_variant_match`
+"""
+
+import sys
+from http import HTTPStatus
+from pprint import pprint
+from typing import TypeVar
+
+import requests
+from dotenv import load_dotenv
+from pydantic import TypeAdapter, ValidationError
+from requests.models import Response
+
+from anyvlm.auth.token_provider import TokenProvider
+from anyvlm.outbound.node_registry import NodeRegistry, VLMNode
+from anyvlm.outbound.variant_match_search import variant_match_search
+from anyvlm.outbound.vlm_client import VLMClient
+from anyvlm.utils.types import (
+    ChromosomeName,
+    GrcAssemblyId,
+    Nucleotide,
+    chromosome_name_adapter,
+    nucleotide_adapter,
+)
+
+load_dotenv()  # required for 'TokenProvider' instantiation
+
+T = TypeVar(name="T")
+
+
+def _exit_early() -> None:
+    """Print an 'exiting' message and exit"""
+    print("🛑 exiting...")
+    sys.exit()
+
+
+def _validate_selected_int(
+    selection: str, max_value: int, min_value: int = 0
+) -> int | None:
+    """Ensure an inputted number is an int, and that it is within the specified range
+
+    :param selection: The selected int to be validated
+    :param max_value: The maximum allowable value for the int
+    :param min_value: [Optional] The minimum allowable value for the int. Defaults to `0` if not set.
+    :returns: The value as an `int` if valid, else `None`
+    """
+    try:
+        selection_number: int = int(selection)
+    except ValueError:
+        return None
+
+    if selection_number <= min_value or selection_number > max_value:
+        return None
+
+    return selection_number
+
+
+def _select_option_helper(
+    options_list: list[T],
+    prompt: str,
+    no_results_message: str,
+    option_attribute: str | None = None,
+) -> T:
+    """A helper function to prompt a user to select one choice from an options list repeatedly until they choose a valid selection
+
+    :param options_list: The list of options from which the user should choose
+    :param prompt: The message to display to the user directing them to make a selection
+    :param no_results_message: The message to display to the user if the 'options_list' is empty
+    :param option_attribute: [Optional] If the options in your 'options_list' are objects, specify the attribute of those objects that should be displayed for the options list
+    :returns: A result of the type specified by your options list
+    """
+    print("\n" + prompt)
+
+    num_options: int = len(options_list)
+    if num_options == 0:
+        print(no_results_message)
+        _exit_early()
+
+    for i, option in enumerate(options_list):
+        selection_value = (
+            getattr(option, option_attribute) if option_attribute else option
+        )
+        print(f" ({i + 1}) {selection_value}")
+
+    while True:
+        selection: str = input("> Enter your selection's number: ")
+        selection_number: int | None = _validate_selected_int(
+            selection=selection, max_value=num_options
+        )
+        if not selection_number:
+            print(f"❌ Please enter a number between 1-{num_options}")
+        else:
+            return options_list[int(selection_number) - 1]
+
+
+def _validate_input_with_type_adapter(
+    type_adapter: TypeAdapter[T], prompt: str, error_message: str | None
+) -> T:
+    """Validates that user inputs matches the constraints defined by the provided TypeAdapter
+
+    :param type_adapter: The TypeAdapter to use for validation
+    :parma prompt: The message to display to the user directing them to make a selection
+    :param error_message: [Optional] A message to display to the user if the validation fails, prompting them to try again. If omitted, the function will fallback to using the error message raised during the failed validation.
+    """
+    print()
+    while True:
+        raw_input: str = input("> " + prompt)
+        try:
+            transformed_input: T = type_adapter.validate_python(raw_input)
+        except ValidationError as e:
+            print(
+                "❌ "
+                + (
+                    error_message
+                    if error_message is not None
+                    else f"{e.errors()[0]['msg']}"
+                )
+            )
+        else:
+            return transformed_input
+
+
+def _select_node(node_registry: NodeRegistry) -> VLMNode:
+    """Prompts the user to select a node to query
+
+    :param node_registry: The list of nodes to choose from
+    :returns: A VLMNode of the user's choice
+    """
+    return _select_option_helper(
+        options_list=[node for node in node_registry.get_nodes() if node.match_url],
+        prompt="Select a node to query from the list below:",
+        no_results_message="Sorry, no nodes are currently registered on the network",
+        option_attribute="client_name",
+    )
+
+
+def _select_assembly() -> GrcAssemblyId:
+    """Prompts the user to select an assembly for the variant they want to match on
+
+    :returns: The user's selected assembly
+    """
+    return _select_option_helper(
+        options_list=list[GrcAssemblyId](GrcAssemblyId),
+        prompt="Select the assembly for the variant you're matching on:",
+        no_results_message="Whoops, there was an error retrieving assembly options",
+    )
+
+
+def _select_reference_name() -> ChromosomeName:
+    """Prompts the user to select a chromosome for the variant they want to match on
+
+    :returns: The user's selected chromosome
+    """
+    return _validate_input_with_type_adapter(
+        type_adapter=chromosome_name_adapter,
+        prompt="Enter the chromosome for the variant you're matching on: ",
+        error_message=None,
+    )
+
+
+def _select_start_position() -> int:
+    """Prompts the user to select a start position for the variant they want to match on
+
+    :returns: The user's selected start position
+    """
+    print()
+    max_start_position = 250000000  # heuristic cap; longest chromosome on either assembly is just under this length
+    while True:
+        start_position_string: str = input(
+            "Enter the start position for the variant you're matching on: "
+        )
+        start_position_number: int | None = _validate_selected_int(
+            selection=start_position_string, max_value=max_start_position
+        )
+        if not start_position_number:
+            print(f"❌ Please enter a number between 1-{max_start_position}")
+        else:
+            return start_position_number
+
+
+def _select_base(base_type: str) -> Nucleotide:
+    """Prompts the user to select a nucleotide base of the specified type
+
+    :param base_type: The type of base to prompt the user for
+    :returns: The user's selected nucleotide base
+    """
+    return _validate_input_with_type_adapter(
+        type_adapter=nucleotide_adapter,
+        prompt=f"Enter the {base_type} base for the variant you're matching on: ",
+        error_message="Please enter a valid nucleotide. (Note: only single bases are supported by the VLM Network at this time)",
+    )
+
+
+def print_args(**kwargs) -> None:  # pyright: ignore[reportUnknownParameterType, reportMissingParameterType]
+    """Prints out the provided arguments to the console"""
+    for name, value in kwargs.items():
+        print(f"\t{name} = {value!r}")
+
+
+def run_variant_match_query() -> None:
+    """Script to prompt a user to query an external VLM Network node"""
+    print("VARIANT MATCH QUERY")
+    vlm_client: VLMClient = VLMClient(token_provider=TokenProvider())
+    node_registry: NodeRegistry = NodeRegistry(vlm_client)
+    selected_node: VLMNode = _select_node(node_registry)
+
+    match_url: str | None = selected_node.match_url
+    if not match_url:
+        print("Sorry, your selected node does not support match queries at this time")
+        _exit_early()
+
+    print("\n⬇️  Tell us more about the variant you're matching against:")
+    assembly: GrcAssemblyId = _select_assembly()
+    reference_name: ChromosomeName = _select_reference_name()
+    start_position: int = _select_start_position()
+    reference_base: Nucleotide = _select_base(base_type="reference")
+    alternate_base: Nucleotide = _select_base(base_type="alternate")
+
+    print("\n---\n")
+    print(
+        f"Querying '{selected_node.client_name}' at '{match_url}' with the following parameters:"
+    )
+    print_args(
+        assembly=assembly.value,
+        reference_name=reference_name,
+        start_position=start_position,
+        reference_base=reference_base,
+        alternate_base=alternate_base,
+    )
+
+    print("\nRESULT:")
+    response: Response | None = None
+    try:
+        response = variant_match_search(
+            query_url=match_url,  # pyright: ignore[reportArgumentType]
+            assembly_id=assembly,
+            reference_name=reference_name,
+            start=start_position,
+            reference_bases=reference_base,
+            alternate_bases=alternate_base,
+        )
+        response.raise_for_status()
+    except requests.exceptions.Timeout:
+        print("❌ ERROR: Request timed out")
+        raise
+    except requests.exceptions.ConnectionError:
+        print("❌ ERROR: Failed to connect to server")
+        raise
+    except requests.exceptions.HTTPError as e:
+        print(f"❌ ERROR: HTTP error: {e.response.status_code}")
+        raise
+    except requests.exceptions.RequestException as e:
+        print(f"❌ ERROR: Request failed: {e}")
+        raise
+
+    if response and response.status_code == HTTPStatus.OK:
+        data = response.json()
+        pprint(object=data)
+    else:
+        print(
+            f"Sorry, there was an error querying this node. We received a response status code of: {response.status_code}"
+        )
+
+
+if __name__ == "__main__":
+    run_variant_match_query()