feat!: add input assembly parameter for free text genomic / gnomad vcf queries (#619)

close #490

Initial work from #491

Technically breaking change with the cleanup work I did

Author: korikuzma

src/variation/main.py

@@ -6,7 +6,7 @@
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from enum import Enum
-from typing import Annotated
+from typing import Annotated, Literal
 from urllib.parse import unquote
 
 import pkg_resources
@@ -40,6 +40,7 @@
     TranslateIdentifierService,
 )
 from variation.schemas.service_schema import (
+    ClinVarAssembly,
     FeatureOverlapService,
     ToCdnaService,
     ToGenomicService,
@@ -176,6 +177,12 @@ async def normalize(
             description="The copy change for HGVS duplications and deletions represented as Copy Number Change Variation.",
         ),
     ] = None,
+    input_assembly: Annotated[
+        Literal[ClinVarAssembly.GRCH37] | Literal[ClinVarAssembly.GRCH38] | None,
+        Query(
+            description="Assembly used for `q`. Only used when `q` is using genomic free text or gnomad vcf format",
+        ),
+    ] = None,
 ) -> NormalizeService:
     """Normalize and translate a HGVS, gnomAD VCF or Free Text description on GRCh37
     or GRCh38 assembly to a single VRS Variation. Performs fully-justified allele
@@ -190,11 +197,14 @@ async def normalize(
     :param copy_change: The copy change for HGVS duplications and deletions represented
         as Copy Number Change Variation. If not set, will use default `copy_change` for
         query.
+    :param input_assembly: Assembly used for `q`. Only used when `q` is using genomic
+        free text or gnomad vcf format
     :return: NormalizeService for variation
     """
     return await query_handler.normalize_handler.normalize(
         unquote(q),
         hgvs_dup_del_mode=hgvs_dup_del_mode,
+        input_assembly=input_assembly,
         baseline_copies=baseline_copies,
         copy_change=copy_change,
     )

src/variation/normalize.py

@@ -1,6 +1,7 @@
 """Module for Variation Normalization."""
 
 import datetime
+from typing import Literal
 from urllib.parse import unquote
 
 from cool_seq_tool.handlers import SeqRepoAccess
@@ -16,6 +17,7 @@
     NormalizeService,
     ServiceMeta,
 )
+from variation.schemas.service_schema import ClinVarAssembly
 from variation.schemas.token_response_schema import GnomadVcfToken, Token
 from variation.schemas.translation_response_schema import (
     AC_PRIORITY_LABELS,
@@ -175,14 +177,21 @@ async def normalize(
         self,
         q: str,
         hgvs_dup_del_mode: HGVSDupDelModeOption | None = HGVSDupDelModeOption.DEFAULT,
+        input_assembly: Literal[ClinVarAssembly.GRCH37, ClinVarAssembly.GRCH38]
+        | None = None,
         baseline_copies: int | None = None,
         copy_change: models.CopyChange | None = None,
     ) -> NormalizeService:
-        """Normalize a given variation.
+        """Normalize and translate a HGVS, gnomAD VCF or Free Text description on GRCh37
+        or GRCh38 assembly to a VRS variation. Performs fully-justfied allele
+        normalization. Will liftover to GRCh38 (if necessary) and align to a priority
+        transcript. Will make inferences about the query.
 
         :param q: HGVS, gnomAD VCF or Free Text description on GRCh37 or GRCh38 assembly
         :param hgvs_dup_del_mode: This parameter determines how to interpret HGVS
             dup/del expressions in VRS.
+        :param input_assembly: Assembly used for `q`. Only used when `q` is using
+            genomic free text or gnomad vcf format
         :param baseline_copies: Baseline copies for HGVS duplications and deletions
         :param copy_change: The copy change for HGVS duplications and deletions
             represented as Copy Number Change Variation.
@@ -226,7 +235,9 @@ async def normalize(
             return NormalizeService(**params)
 
         # Get validation summary for classification
-        validation_summary = await self.validator.perform(classification)
+        validation_summary = await self.validator.perform(
+            classification, input_assembly=input_assembly
+        )
         if not validation_summary:
             update_warnings_for_no_resp(label, validation_summary.warnings)
             params["warnings"] = warnings

src/variation/translators/translator.py

@@ -19,7 +19,6 @@
     VrsSeqLocAcStatus,
 )
 from variation.schemas.validation_response_schema import ValidationResult
-from variation.validators.genomic_base import GenomicBase
 from variation.vrs_representation import VRSRepresentation
 
 
@@ -44,7 +43,6 @@ def __init__(
         """
         self.seqrepo_access = seqrepo_access
         self.uta = uta
-        self.genomic_base = GenomicBase(self.seqrepo_access, self.uta)
         self.mane_transcript = mane_transcript
         self.vrs = vrs
         self.hgvs_dup_del_mode = hgvs_dup_del_mode

src/variation/utils.py

@@ -208,8 +208,8 @@ def get_refget_accession(
     else:
         if not ids:
             errors.append(f"Unable to find ga4gh sequence identifiers for: {alias}")
-
-        refget_accession = ids[0].split("ga4gh:")[-1]
+        else:
+            refget_accession = ids[0].split("ga4gh:")[-1]
     return refget_accession
 
 

src/variation/validate.py

@@ -1,11 +1,14 @@
 """Module for Validation."""
 
+from typing import Literal
+
 from cool_seq_tool.handlers import SeqRepoAccess
 from cool_seq_tool.mappers import LiftOver
 from cool_seq_tool.sources import TranscriptMappings, UtaDatabase
 from gene.query import QueryHandler as GeneQueryHandler
 
 from variation.schemas.classification_response_schema import Classification
+from variation.schemas.service_schema import ClinVarAssembly
 from variation.schemas.validation_response_schema import ValidationSummary
 from variation.validators import (
     Amplification,
@@ -29,7 +32,7 @@
     ProteinStopGain,
     ProteinSubstitution,
 )
-from variation.validators.validator import Validator
+from variation.validators.validator import GenomicValidator, Validator
 
 
 class Validate:
@@ -76,11 +79,18 @@ def __init__(
             Amplification(*params),
         ]
 
-    async def perform(self, classification: Classification) -> ValidationSummary:
+    async def perform(
+        self,
+        classification: Classification,
+        input_assembly: Literal[ClinVarAssembly.GRCH37, ClinVarAssembly.GRCH38]
+        | None = None,
+    ) -> ValidationSummary:
         """Get validation summary containing invalid and valid results for a
         classification
 
         :param classification: A classification for a list of tokens
+        :param input_assembly: Assembly used for `q`. Only used when `q` is using
+            genomic free text of gnomad vcf format
         :return: Validation summary for classification containing valid and invalid
             results
         """
@@ -94,7 +104,15 @@ async def perform(self, classification: Classification) -> ValidationSummary:
             if validator.validates_classification_type(
                 classification.classification_type
             ):
-                validation_results = await validator.validate(classification)
+                if isinstance(validator, GenomicValidator):
+                    validation_results = await validator.validate(
+                        classification, input_assembly=input_assembly
+                    )
+                else:
+                    validation_results = await validator.validate(
+                        classification,
+                    )
+
                 for validation_result in validation_results:
                     if validation_result.is_valid:
                         found_valid_result = True

src/variation/validators/__init__.py

@@ -6,7 +6,6 @@
 from .cdna_insertion import CdnaInsertion
 from .cdna_reference_agree import CdnaReferenceAgree
 from .cdna_substitution import CdnaSubstitution
-from .genomic_base import GenomicBase
 from .genomic_deletion import GenomicDeletion
 from .genomic_deletion_ambiguous import GenomicDeletionAmbiguous
 from .genomic_delins import GenomicDelIns
@@ -29,7 +28,6 @@
     "CdnaInsertion",
     "CdnaReferenceAgree",
     "CdnaSubstitution",
-    "GenomicBase",
     "GenomicDelIns",
     "GenomicDeletion",
     "GenomicDeletionAmbiguous",

src/variation/validators/genomic_base.py

@@ -1,16 +1,15 @@
 """The module for Genomic Deletion Validation."""
 
 from variation.schemas.classification_response_schema import (
-    Classification,
     ClassificationType,
     GenomicDeletionClassification,
     Nomenclature,
 )
 from variation.schemas.validation_response_schema import ValidationResult
-from variation.validators.validator import Validator
+from variation.validators.validator import GenomicValidator
 
 
-class GenomicDeletion(Validator):
+class GenomicDeletion(GenomicValidator):
     """The Genomic Deletion Validator class."""
 
     async def get_valid_invalid_results(
@@ -108,21 +107,3 @@ def validates_classification_type(
     ) -> bool:
         """Return whether or not the classification type is genomic deletion"""
         return classification_type == ClassificationType.GENOMIC_DELETION
-
-    async def get_accessions(
-        self, classification: Classification, errors: list
-    ) -> list[str]:
-        """Get accessions for a given classification.
-        If `classification.nomenclature == Nomenclature.HGVS`, will return the accession
-        in the HGVS expression.
-        Else, will get all accessions associated to the gene
-
-        :param classification: The classification for list of tokens
-        :param errors: List of errors
-        :return: List of accessions
-        """
-        if classification.nomenclature == Nomenclature.HGVS:
-            accessions = [classification.ac]
-        else:
-            accessions = await self.get_genomic_accessions(classification, errors)
-        return accessions

src/variation/validators/genomic_deletion.py

@@ -2,16 +2,14 @@
 
 from variation.schemas.classification_response_schema import (
     AmbiguousType,
-    Classification,
     ClassificationType,
     GenomicDeletionAmbiguousClassification,
-    Nomenclature,
 )
 from variation.schemas.validation_response_schema import ValidationResult
-from variation.validators.validator import Validator
+from variation.validators.validator import GenomicValidator
 
 
-class GenomicDeletionAmbiguous(Validator):
+class GenomicDeletionAmbiguous(GenomicValidator):
     """The Genomic Deletion Ambiguous Validator class."""
 
     async def get_valid_invalid_results(
@@ -100,21 +98,3 @@ def validates_classification_type(
         ambiguous
         """
         return classification_type == ClassificationType.GENOMIC_DELETION_AMBIGUOUS
-
-    async def get_accessions(
-        self, classification: Classification, errors: list
-    ) -> list[str]:
-        """Get accessions for a given classification.
-        If `classification.nomenclature == Nomenclature.HGVS`, will return the accession
-        in the HGVS expression.
-        Else, will get all accessions associated to the gene
-
-        :param classification: The classification for list of tokens
-        :param errors: List of errors
-        :return: List of accessions
-        """
-        if classification.nomenclature == Nomenclature.HGVS:
-            accessions = [classification.ac]
-        else:
-            accessions = await self.get_genomic_accessions(classification, errors)
-        return accessions

src/variation/validators/genomic_deletion_ambiguous.py

@@ -1,16 +1,15 @@
 """The module for Genomic DelIns Validation."""
 
 from variation.schemas.classification_response_schema import (
-    Classification,
     ClassificationType,
     GenomicDelInsClassification,
     Nomenclature,
 )
 from variation.schemas.validation_response_schema import ValidationResult
-from variation.validators.validator import Validator
+from variation.validators.validator import GenomicValidator
 
 
-class GenomicDelIns(Validator):
+class GenomicDelIns(GenomicValidator):
     """The Genomic DelIns Validator class."""
 
     async def get_valid_invalid_results(
@@ -79,21 +78,3 @@ def validates_classification_type(
     ) -> bool:
         """Return whether or not the classification type is genomic delins"""
         return classification_type == ClassificationType.GENOMIC_DELINS
-
-    async def get_accessions(
-        self, classification: Classification, errors: list
-    ) -> list[str]:
-        """Get accessions for a given classification.
-        If `classification.nomenclature == Nomenclature.HGVS`, will return the accession
-        in the HGVS expression.
-        Else, will get all accessions associated to the gene
-
-        :param classification: The classification for list of tokens
-        :param errors: List of errors
-        :return: List of accessions
-        """
-        if classification.nomenclature == Nomenclature.HGVS:
-            accessions = [classification.ac]
-        else:
-            accessions = await self.get_genomic_accessions(classification, errors)
-        return accessions