create AnyVlmCohortAlleleFrequencyResult

Author: korikuzma

pyproject.toml

@@ -18,7 +18,7 @@ classifiers = [
 ]
 dependencies = [
     "ga4gh.vrs>=2.2.0,<3.0",
-    "ga4gh.va_spec~=0.4.2",
+    "ga4gh.va_spec~=0.4.3",
     "biocommons.anyvar@git+https://github.com/biocommons/anyvar.git@0d3ab56fe936b27235a1ce136da4641ea81c0bbf",
     "fastapi>=0.95.0",
     "python-multipart",  # required for fastapi file uploads

src/anyvlm/anyvar/http_client.py

@@ -100,15 +100,15 @@ def _make_allele_expression_request(
     def get_registered_allele_expression(
         self, expression: str, assembly: ReferenceAssembly = ReferenceAssembly.GRCH38
     ) -> models.Allele | None:
-        """Retrieve VRS Allele for given allele expression
+        """Retrieve registered VRS Allele for given allele expression
 
         Currently, only expressions supported by the VRS-Python translator are supported.
         This could change depending on the AnyVar implementation, though, and probably
         can't be validated on the AnyVLM side.
 
-        :param expression: variation expression to translate
+        :param expression: variation expression to get VRS Allele for
         :param assembly: reference assembly used in expression
-        :return: VRS Allele if translation succeeds, else `None`
+        :return: VRS Allele if translation succeeds and VRS Allele has already been registered, else `None`
         """
         response = self._make_allele_expression_request(
             expression, assembly, HTTPMethod.POST

src/anyvlm/anyvar/python_client.py

@@ -30,7 +30,7 @@ def __init__(self, translator: Translator, storage: Storage) -> None:
     def _translate_allele_expression(
         self, expression: str, assembly: ReferenceAssembly = ReferenceAssembly.GRCH38
     ) -> Allele | None:
-        """Translate a single allele expression to a VRS Allele ID
+        """Translate a single allele expression to a VRS Allele
 
         Currently, only expressions supported by the VRS-Python translator are supported.
         This could change depending on the AnyVar implementation, though, and probably
@@ -67,14 +67,15 @@ def get_registered_allele_expression(
         :return: VRS Allele if translation succeeds and VRS Allele has already been registered, else `None`
         """
         translated_variation = self._translate_allele_expression(expression, assembly)
-        if translated_variation:
-            try:
-                return self.av.get_object(translated_variation.id, Allele)  # type: ignore
-            except KeyError:
-                _logger.exception(
-                    "VRS Allele with ID %s not found", translated_variation.id
-                )
-        return None
+        if not translated_variation:
+            return None
+
+        try:
+            return self.av.get_object(translated_variation.id, Allele)  # type: ignore
+        except KeyError:
+            _logger.exception(
+                "VRS Allele with ID %s not found", translated_variation.id
+            )
 
     def put_allele_expressions(
         self,

src/anyvlm/functions/build_vlm_response.py

@@ -1,18 +1,18 @@
 """Craft a VlmResponse object from a list of CohortAlleleFrequencyStudyResults"""
 
-from ga4gh.va_spec.base.core import CohortAlleleFrequencyStudyResult
+from ga4gh.va_spec.base.core import AnyVlmCohortAlleleFrequencyResult
 
 from anyvlm.schemas.vlm import (
     VlmResponse,
 )
 
 
 def build_vlm_response_from_caf_data(
-    caf_data: list[CohortAlleleFrequencyStudyResult],
+    caf_data: list[AnyVlmCohortAlleleFrequencyResult],
 ) -> VlmResponse:
     """Craft a VlmResponse object from a list of CohortAlleleFrequencyStudyResults.
 
-    :param caf_data: A list of `CohortAlleleFrequencyStudyResult` objects that will be used to build the VlmResponse
+    :param caf_data: A list of `AnyVlmCohortAlleleFrequencyResult` objects that will be used to build the VlmResponse
     :return: A `VlmResponse` object.
     """
     raise NotImplementedError  # TODO: Implement this during/after Issue #16

src/anyvlm/functions/get_caf.py

@@ -1,12 +1,12 @@
 """Perform search against variant(s) contained by an AnyVar node, and construct cohort allele frequency model(s)"""
 
 from ga4gh.core.models import iriReference
-from ga4gh.va_spec.base import CohortAlleleFrequencyStudyResult
 
 from anyvlm.anyvar.base_client import BaseAnyVarClient
 from anyvlm.storage.base_storage import Storage
 from anyvlm.utils.types import (
     ASSEMBLY_MAP,
+    AnyVlmCohortAlleleFrequencyResult,
     ChromosomeName,
     GrcAssemblyId,
     NucleotideSequence,
@@ -26,7 +26,7 @@ def get_caf(
     start: int,
     reference_bases: NucleotideSequence,
     alternate_bases: NucleotideSequence,
-) -> list[CohortAlleleFrequencyStudyResult]:
+) -> list[AnyVlmCohortAlleleFrequencyResult]:
     """Retrieve Cohort Allele Frequency data for all known registered variants matching
     provided search params
 
@@ -41,7 +41,7 @@ def get_caf(
     :param alternate_bases: Genomic bases ('T', 'AC', etc.)
     :raises ValueError: if unsupported assembly ID is provided
     :raises VariantNotRegisteredError: if variant is not registered in AnyVar
-    :return: list of CohortAlleleFrequencyStudyResult objects for the provided variant
+    :return: list of AnyVlmCohortAlleleFrequencyResult objects for the provided variant
     """
     gnomad_vcf: str = f"{reference_name}-{start}-{reference_bases}-{alternate_bases}"
     try:
@@ -55,7 +55,7 @@ def get_caf(
         msg = f"Variant {assembly.value} {gnomad_vcf} is not registered in AnyVar"
         raise VariantNotRegisteredError(msg)
 
-    cafs: list[CohortAlleleFrequencyStudyResult] = (
+    cafs: list[AnyVlmCohortAlleleFrequencyResult] = (
         anyvlm_storage.get_caf_by_vrs_allele_id(vrs_variation.id)  # type: ignore
     )
 

src/anyvlm/restapi/vlm.py

@@ -4,7 +4,7 @@
 from typing import Annotated
 
 from fastapi import Query, Request
-from ga4gh.va_spec.base.core import CohortAlleleFrequencyStudyResult
+from ga4gh.va_spec.base.core import AnyVlmCohortAlleleFrequencyResult
 
 from anyvlm.anyvar.base_client import BaseAnyVarClient
 from anyvlm.functions.build_vlm_response import build_vlm_response_from_caf_data
@@ -57,7 +57,7 @@ def variant_counts(
 ) -> VlmResponse:
     anyvar_client: BaseAnyVarClient = request.app.state.anyvar_client
     anyvlm_storage: Storage = request.app.state.anyvlm_storage
-    caf_data: list[CohortAlleleFrequencyStudyResult] = get_caf(
+    caf_data: list[AnyVlmCohortAlleleFrequencyResult] = get_caf(
         anyvar_client,
         anyvlm_storage,
         assemblyId,

src/anyvlm/storage/base_storage.py

@@ -2,7 +2,7 @@
 
 from abc import ABC, abstractmethod
 
-from ga4gh.va_spec.base import CohortAlleleFrequencyStudyResult
+from anyvlm.utils.types import AnyVlmCohortAlleleFrequencyResult
 
 
 class StorageError(Exception):
@@ -30,7 +30,7 @@ def sanitized_url(self) -> str:
         """Return a sanitized URL (password masked) of the database connection string."""
 
     @abstractmethod
-    def add_allele_frequencies(self, caf: CohortAlleleFrequencyStudyResult) -> None:
+    def add_allele_frequencies(self, caf: AnyVlmCohortAlleleFrequencyResult) -> None:
         """Add allele frequency data to the database. Will skip conflicts.
 
         NOTE: For now, this will only insert a single caf record into the database.
@@ -43,10 +43,10 @@ def add_allele_frequencies(self, caf: CohortAlleleFrequencyStudyResult) -> None:
     @abstractmethod
     def get_caf_by_vrs_allele_id(
         self, vrs_allele_id: str
-    ) -> list[CohortAlleleFrequencyStudyResult]:
+    ) -> list[AnyVlmCohortAlleleFrequencyResult]:
         """Retrieve cohort allele frequency study results by VRS Allele ID
 
-        :param vrs_allele_id: VRS Allele ID
-        :return: List of cohort allele frequency study results matching given VRS
-            Allele IDs. Will use iriReference for focusAllele
+        :param vrs_allele_id: VRS Allele ID to filter by
+        :return: List of cohort allele frequency study results matching given VRS Allele
+            ID. Will use iriReference for focusAllele
         """

src/anyvlm/storage/mapper_registry.py

@@ -3,10 +3,9 @@
 from types import MappingProxyType
 from typing import TypeVar
 
-from ga4gh.va_spec.base import CohortAlleleFrequencyStudyResult
-
 from anyvlm.storage import orm
 from anyvlm.storage.mappers import AlleleFrequencyMapper, BaseMapper
+from anyvlm.utils.types import AnyVlmCohortAlleleFrequencyResult
 
 T = TypeVar("T")
 
@@ -15,7 +14,7 @@ class MapperRegistry:
     """Central registry for all object mappers."""
 
     va_model_to_db_mapping: MappingProxyType = MappingProxyType(
-        {CohortAlleleFrequencyStudyResult: orm.AlleleFrequencyData}
+        {AnyVlmCohortAlleleFrequencyResult: orm.AlleleFrequencyData}
     )
 
     _mappers: MappingProxyType[type, BaseMapper] = MappingProxyType(

src/anyvlm/storage/mappers.py

@@ -4,13 +4,16 @@
 from typing import Generic, TypeVar
 
 from ga4gh.core.models import iriReference
-from ga4gh.va_spec.base import CohortAlleleFrequencyStudyResult, StudyGroup
-from pydantic import ValidationError
+from ga4gh.va_spec.base import StudyGroup
 
 from anyvlm.storage import orm
-from anyvlm.utils.types import AncillaryResults, QualityMeasures
+from anyvlm.utils.types import (
+    AncillaryResults,
+    AnyVlmCohortAlleleFrequencyResult,
+    QualityMeasures,
+)
 
-V = TypeVar("V")  # VA-Spec entity type
+V = TypeVar("V")  # VA-Spec compliant entity type
 D = TypeVar("D")  # DB entity type
 
 
@@ -19,73 +22,77 @@ class BaseMapper(Generic[V, D], ABC):
 
     @abstractmethod
     def from_db_entity(self, db_entity: D) -> V:
-        """Convert DB entity to VA-Spec model."""
+        """Convert DB entity to VA-Spec compliant model."""
 
     @abstractmethod
     def to_db_entity(self, va_model: V) -> D:
-        """Convert VA-Spec model to DB entity."""
+        """Convert VA-Spec compliant model to DB entity."""
 
 
 class AlleleFrequencyMapper(
-    BaseMapper[CohortAlleleFrequencyStudyResult, orm.AlleleFrequencyData]
+    BaseMapper[AnyVlmCohortAlleleFrequencyResult, orm.AlleleFrequencyData]
 ):
     """Maps between Allele Frequency Entities"""
 
     def from_db_entity(
         self, db_entity: orm.AlleleFrequencyData
-    ) -> CohortAlleleFrequencyStudyResult:
+    ) -> AnyVlmCohortAlleleFrequencyResult:
         """Convert DB Allele Frequency Data to VA-Spec Cohort Allele Frequency Study Result model
 
         :param db_entity: An ORM Allele Frequency Data instance
-        :return: VA-Spec Cohort Allele Frequency Study Result instance. Will use
-            iriReference for focusAllele
+        :return: VA-Spec compliant Cohort Allele Frequency Study Result instance. Will
+            use iriReference for focusAllele
         """
         homozygotes = db_entity.ac_hom
         heterozygotes = db_entity.ac_het
         hemizygotes = db_entity.ac_hemi
-        ac = sum(x or 0 for x in (homozygotes, heterozygotes, hemizygotes))
-        an = db_entity.an
 
-        if filter_ := db_entity.filter:
-            quality_measures = QualityMeasures(qcFilters=filter_).model_dump()
+        if any(x is not None for x in (homozygotes, heterozygotes, hemizygotes)):
+            ancillary_results = AncillaryResults(
+                homozygotes=homozygotes,
+                heterozygotes=heterozygotes,
+                hemizygotes=hemizygotes,
+            )
         else:
-            quality_measures = None
+            ancillary_results = None
 
-        return CohortAlleleFrequencyStudyResult(
+        ac = sum(x or 0 for x in (homozygotes, heterozygotes, hemizygotes))
+        an = db_entity.an
+
+        return AnyVlmCohortAlleleFrequencyResult(
             focusAllele=iriReference(db_entity.vrs_id),
             focusAlleleCount=ac,
             locusAlleleCount=an,
             focusAlleleFrequency=round(ac / an, 9),
-            qualityMeasures=quality_measures,
-            ancillaryResults=AncillaryResults(
-                homozygotes=homozygotes,
-                heterozygotes=heterozygotes,
-                hemizygotes=hemizygotes,
-            ).model_dump(),
+            qualityMeasures=QualityMeasures(qcFilters=db_entity.filter)
+            if db_entity.filter
+            else None,
+            ancillaryResults=ancillary_results,
             cohort=StudyGroup(name=db_entity.cohort),  # type: ignore
         )
 
     def to_db_entity(
-        self, va_model: CohortAlleleFrequencyStudyResult
+        self, va_model: AnyVlmCohortAlleleFrequencyResult
     ) -> orm.AlleleFrequencyData:
-        """Convert VA-Spec Cohort Allele Frequency Study Result model to DB Allele Frequency Data
+        """Convert VA-Spec compliant Cohort Allele Frequency Study Result model to DB
+        Allele Frequency Data
 
-        :param va_model: VA-Spec Cohort Allele Frequency Study Result instance
+        :param va_model: VA-Spec compliant Cohort Allele Frequency Study Result instance
         :return: ORM Allele Frequency Data instance
-        :raises ValueError: if ancillaryResults or qualityMeasures are invalid
         """
-        try:
-            ancillary_results = AncillaryResults(**va_model.ancillaryResults or {})
-        except ValidationError as e:
-            raise ValueError("Invalid ancillaryResults data") from e
+        ancillary_results = va_model.ancillaryResults
+        if ancillary_results:
+            ac_het = ancillary_results.heterozygotes
+            ac_hom = ancillary_results.homozygotes
+            ac_hemi = ancillary_results.hemizygotes
+        else:
+            ac_het = None
+            ac_hom = None
+            ac_hemi = None
 
-        try:
-            quality_measures = QualityMeasures(**va_model.qualityMeasures or {})
-        except ValidationError as e:
-            raise ValueError("Invalid qualityMeasures data") from e
+        quality_measures = va_model.qualityMeasures
 
         focus_allele = va_model.focusAllele
-
         if isinstance(focus_allele, iriReference):
             vrs_id = focus_allele.root
         else:
@@ -95,9 +102,9 @@ def to_db_entity(
             vrs_id=vrs_id,
             an=va_model.locusAlleleCount,
             ac=va_model.focusAlleleCount,
-            ac_het=ancillary_results.heterozygotes,
-            ac_hom=ancillary_results.homozygotes,
-            ac_hemi=ancillary_results.hemizygotes,
-            filter=quality_measures.qcFilters,
+            ac_het=ac_het,
+            ac_hom=ac_hom,
+            ac_hemi=ac_hemi,
+            filter=quality_measures.qcFilters if quality_measures else None,
             cohort=va_model.cohort.name,
         )

src/anyvlm/storage/postgres.py

@@ -2,7 +2,6 @@
 
 from urllib.parse import urlparse
 
-from ga4gh.va_spec.base import CohortAlleleFrequencyStudyResult
 from sqlalchemy import create_engine, delete, select
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.orm import sessionmaker
@@ -13,6 +12,7 @@
 )
 from anyvlm.storage.mapper_registry import mapper_registry
 from anyvlm.storage.orm import create_tables
+from anyvlm.utils.types import AnyVlmCohortAlleleFrequencyResult
 
 
 class PostgresObjectStore(Storage):
@@ -54,7 +54,7 @@ def sanitized_url(self) -> str:
             netloc += f":{parsed.port}"
         return f"{parsed.scheme}://{netloc}{parsed.path}"
 
-    def add_allele_frequencies(self, caf: CohortAlleleFrequencyStudyResult) -> None:
+    def add_allele_frequencies(self, caf: AnyVlmCohortAlleleFrequencyResult) -> None:
         """Add allele frequency data to the database. Will skip conflicts.
 
         NOTE: For now, this will only insert a single caf record into the database.
@@ -71,14 +71,14 @@ def add_allele_frequencies(self, caf: CohortAlleleFrequencyStudyResult) -> None:
 
     def get_caf_by_vrs_allele_id(
         self, vrs_allele_id: str
-    ) -> list[CohortAlleleFrequencyStudyResult]:
+    ) -> list[AnyVlmCohortAlleleFrequencyResult]:
         """Retrieve cohort allele frequency study results by VRS Allele ID
 
-        :param vrs_allele_id: VRS Allele ID
-        :return: List of cohort allele frequency study results matching given VRS
-            Allele ID. Will use iriReference for focusAllele
+        :param vrs_allele_id: VRS Allele ID to filter by
+        :return: List of cohort allele frequency study results matching given VRS Allele
+            ID. Will use iriReference for focusAllele
         """
-        cafs: list[CohortAlleleFrequencyStudyResult] = []
+        cafs: list[AnyVlmCohortAlleleFrequencyResult] = []
         with self.session_factory() as session:
             stmt = (
                 select(orm.AlleleFrequencyData)