diff --git a/pyproject.toml b/pyproject.toml index 5c078df..a36d2b6 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/anyvlm/outbound/node_registry.py b/src/anyvlm/outbound/node_registry.py index b29cd0f..be37ce1 100644 --- a/src/anyvlm/outbound/node_registry.py +++ b/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"), diff --git a/src/anyvlm/outbound/variant_match_search.py b/src/anyvlm/outbound/variant_match_search.py new file mode 100644 index 0000000..683eea6 --- /dev/null +++ b/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) diff --git a/src/anyvlm/scripts/__init__.py b/src/anyvlm/scripts/__init__.py new file mode 100644 index 0000000..e4aa5db --- /dev/null +++ b/src/anyvlm/scripts/__init__.py @@ -0,0 +1 @@ +"""Utility scripts""" diff --git a/src/anyvlm/scripts/run_variant_match.py b/src/anyvlm/scripts/run_variant_match.py new file mode 100644 index 0000000..d334075 --- /dev/null +++ b/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() diff --git a/src/anyvlm/utils/types.py b/src/anyvlm/utils/types.py index 552991a..cd930d4 100644 --- a/src/anyvlm/utils/types.py +++ b/src/anyvlm/utils/types.py @@ -2,11 +2,17 @@ from enum import StrEnum from types import MappingProxyType -from typing import Annotated +from typing import Annotated, TypeAlias from anyvar.mapping.liftover import ReferenceAssembly from ga4gh.va_spec.base import CohortAlleleFrequencyStudyResult -from pydantic import AfterValidator, BaseModel, BeforeValidator, StringConstraints +from pydantic import ( + AfterValidator, + BaseModel, + BeforeValidator, + StringConstraints, + TypeAdapter, +) class AncillaryResults(BaseModel): @@ -67,11 +73,12 @@ class UcscAssemblyBuild(StrEnum): ) -Nucleotide = Annotated[ +Nucleotide: TypeAlias = Annotated[ str, BeforeValidator(str.upper), StringConstraints(pattern=r"^[ACGT]$"), ] +nucleotide_adapter: TypeAdapter[Nucleotide] = TypeAdapter[str](Nucleotide) def _normalize_chromosome_name(chromosome_name: str) -> str: @@ -98,7 +105,8 @@ def _normalize_chromosome_name(chromosome_name: str) -> str: ) -ChromosomeName = Annotated[str, AfterValidator(_normalize_chromosome_name)] +ChromosomeName: TypeAlias = Annotated[str, AfterValidator(_normalize_chromosome_name)] +chromosome_name_adapter: TypeAdapter[ChromosomeName] = TypeAdapter[str](ChromosomeName) class Zygosity(StrEnum):