add python api

Author: jsstevenson

src/anyvlm/anyvar/base_client.py

@@ -9,10 +9,6 @@ class AnyVarClientError(Exception):
     """Generic client-related exception."""
 
 
-class AnyVarConnectionError(AnyVarClientError):
-    """Raise for network/communication failures when making calls to AnyVar instance"""
-
-
 class BaseAnyVarClient(abc.ABC):
     """Interface elements for an AnyVar client"""
 
@@ -22,7 +18,7 @@ def put_objects(self, objects: list[VrsVariation]) -> list[VrsVariation]:
 
         :param objects: variation objects to register
         :return: completed VRS objects
-        :raise AnyVarConnectionError: if connection is unsuccessful during registration request
+        :raise AnyVarClientError: for errors relating to specifics of client interface
         """
 
     @abc.abstractmethod
@@ -35,7 +31,7 @@ def search_by_interval(
         :param start: start position for genomic region
         :param end: end position for genomic region
         :return: list of matching variant objects
-        :raise AnyVarConnectionError: if connection is unsuccessful during search query
+        :raise AnyVarClientError: if connection is unsuccessful during search query
         """
 
     @abc.abstractmethod

src/anyvlm/anyvar/http_client.py

@@ -4,7 +4,7 @@
 from anyvar.utils.types import VrsVariation
 from ga4gh.vrs import models
 
-from anyvlm.anyvar.base_client import AnyVarConnectionError, BaseAnyVarClient
+from anyvlm.anyvar.base_client import AnyVarClientError, BaseAnyVarClient
 
 
 class HttpAnyVarClient(BaseAnyVarClient):
@@ -30,20 +30,19 @@ def put_objects(self, objects: list[VrsVariation]) -> list[VrsVariation]:
 
         :param objects: variation objects to register
         :return: completed VRS objects
-        :raise AnyVarConnectionError: if connection is unsuccessful during registration request
+        :raise AnyVarClientError: if connection is unsuccessful during registration request
         """
         results = []
-        url = f"{self.hostname}/vrs_variation"
         for vrs_object in objects:
             response = requests.put(
-                url,
+                f"{self.hostname}/vrs_variation",
                 json=vrs_object.model_dump(exclude_none=True, mode="json"),
                 timeout=self.request_timeout,
             )
             try:
                 response.raise_for_status()
             except requests.HTTPError as e:
-                raise AnyVarConnectionError from e
+                raise AnyVarClientError from e
             result_object = response.json()["object"]
             if result_object.get("type") == "Allele":
                 results.append(models.Allele(**result_object))
@@ -62,7 +61,7 @@ def search_by_interval(
         :param start: start position for genomic region
         :param end: end position for genomic region
         :return: list of matching variant objects
-        :raise AnyVarConnectionError: if connection is unsuccessful during search query
+        :raise AnyVarClientError: if connection is unsuccessful during search query
         """
         response = requests.get(
             f"{self.hostname}/search?accession={accession}&start={start}&end={end}",
@@ -71,7 +70,11 @@ def search_by_interval(
         try:
             response.raise_for_status()
         except requests.HTTPError as e:
-            raise AnyVarConnectionError from e
+            if response.json() == {
+                "detail": "Unable to dereference provided accession ID"
+            }:
+                return []
+            raise AnyVarClientError from e
         return [models.Allele(**v) for v in response.json()["variations"]]
 
     def close(self) -> None:

src/anyvlm/anyvar/python_client.py

@@ -0,0 +1,60 @@
+"""Implement AnyVar client interface for direct Python-based access."""
+
+from anyvar import AnyVar
+from anyvar.storage.base_storage import IncompleteVrsObjectError, Storage
+from anyvar.translate.translate import Translator
+from anyvar.utils.types import VrsVariation, recursive_identify
+
+from anyvlm.anyvar.base_client import BaseAnyVarClient
+
+
+class PythonAnyVarClient(BaseAnyVarClient):
+    """A Python-based AnyVar client."""
+
+    def __init__(self, translator: Translator, storage: Storage) -> None:  # noqa: D107
+        self.av = AnyVar(translator, storage)
+
+    def put_objects(self, objects: list[VrsVariation]) -> list[VrsVariation]:
+        """Register objects with AnyVar
+
+        :param objects: variation objects to register
+        :return: completed VRS objects
+        """
+        complete_variations = []
+        for variation in objects:
+            try:
+                self.av.put_objects([variation])
+            except IncompleteVrsObjectError:
+                variation = recursive_identify(variation)  # noqa: PLW2901
+                self.av.put_objects([variation])
+            complete_variations.append(variation)
+        return complete_variations
+
+    def search_by_interval(
+        self, accession: str, start: int, end: int
+    ) -> list[VrsVariation]:
+        """Get all variation IDs located within the specified range
+
+        :param accession: sequence accession
+        :param start: start position for genomic region
+        :param end: end position for genomic region
+        :return: list of matching variant objects
+        """
+        try:
+            if accession.startswith("ga4gh:"):
+                ga4gh_id = accession
+            else:
+                ga4gh_id = self.av.translator.get_sequence_id(accession)
+        except KeyError:
+            return []
+
+        alleles = []
+        if ga4gh_id:
+            refget_accession = ga4gh_id.split("ga4gh:")[-1]
+            alleles = self.av.object_store.search_alleles(refget_accession, start, end)
+
+        return alleles  # type: ignore[reportReturnType]
+
+    def close(self) -> None:
+        """Clean up AnyVar instance."""
+        self.av.object_store.close()