feat(gks): add support for vrs alleles

close #20

* In addition to Categorical Variants, VRS Alleles will be recognized

Author: korikuzma

clinvar_this/io/gks_json/base.py

@@ -26,7 +26,7 @@
     VariantOncogenicityProposition,
 )
 from ga4gh.va_spec.ccv_2022 import VariantOncogenicityStatement
-from ga4gh.vrs.models import Expression, Syntax
+from ga4gh.vrs.models import Allele, Expression, MolecularVariation, Syntax
 from pydantic import BaseModel, ConfigDict
 
 
@@ -185,19 +185,21 @@ def _read_file(
 
     @staticmethod
     def _get_variant_hgvs(
-        variant: CategoricalVariant,
+        variant: CategoricalVariant | Allele,
     ) -> str | None:
         """Retrieve a HGVS expression for a variant
 
-        Checks the first constraint for an expression. Only support
-        DefiningAlleleConstraints at the moment.
+        For Categorical Variants, checks the first constraint for an expression. Only
+        support extracting from DefiningAlleleConstraints at the moment.
+        If no constraints found, then checks the expressions extension. These are cases
+        where an HGVS expression is unable to be representing using VRS.
 
-        If no constraints found, then checks the expressions extension.
+        For VRS Alleles, checks the `expressions` field.
 
         Order matters: the first matching expression is returned. cDNA RefSeq HGVS
         expressions are prioritized over genomic RefSeq HGVS expressions.
 
-        :param variant: Categorical Variant
+        :param variant: Variant associated to statement
         :return: cDNA RefSeq HGVS expression or genomic RefSeq HGVS expression for a
             variant, if provided.
         """
@@ -225,20 +227,24 @@ def get_hgvs(
             return None
 
         expressions = None
-        if getattr(variant, "constraints", None) and variant.constraints:
-            constraint = variant.constraints[0]
-            if isinstance(constraint.root, DefiningAlleleConstraint):
-                expressions = constraint.root.allele.expressions
+
+        if isinstance(variant, CategoricalVariant):
+            if getattr(variant, "constraints", None) and variant.constraints:
+                constraint = variant.constraints[0]
+                if isinstance(constraint.root, DefiningAlleleConstraint):
+                    expressions = constraint.root.allele.expressions
+            else:
+                # Case where a VRS Allele is unable to be represented, can store
+                # expressions as an extension named 'expressions'
+                try:
+                    expressions_ext = next(
+                        ext for ext in variant.extensions if ext.name == "expressions"
+                    ).value
+                    expressions = [Expression(**ext) for ext in expressions_ext]
+                except (StopIteration, TypeError):
+                    return None
         else:
-            # Case where a VRS Allele is unable to be represented, can store
-            # expressions as an extension named 'expressions'
-            try:
-                expressions_ext = next(
-                    ext for ext in variant.extensions if ext.name == "expressions"
-                ).value
-                expressions = [Expression(**ext) for ext in expressions_ext]
-            except (StopIteration, TypeError):
-                return None
+            expressions = variant.expressions
 
         return get_hgvs(expressions, Syntax.HGVS_C) or get_hgvs(
             expressions, Syntax.HGVS_G
@@ -518,29 +524,25 @@ def _get_clinvar_accession_and_record_status(statement: GksStatementT) -> dict:
 
     def _get_variant_set(
         self,
-        proposition: VariantTherapeuticResponseProposition
-        | VariantDiagnosticProposition
-        | VariantPrognosticProposition,
+        gene_context: MappableConcept,
+        variant: CategoricalVariant | Allele,
         variant_hgvs: str | None = None,
     ) -> SubmissionVariantSet:
         """Get variant set
 
         This assumes only a single submission variant.
 
-        :param proposition: Proposition for a given statement.
+        :param gene_context: Gene associated to statement
+        :param variant: Variant associated to statement
         :param variant_hgvs: The HGVS expression for a variant, if found.
         :return: Variant set for a proposition
         """
         return SubmissionVariantSet(
             variant=[
                 SubmissionVariant(
                     hgvs=variant_hgvs,
-                    gene=[
-                        SubmissionVariantGene(
-                            symbol=proposition.geneContextQualifier.name
-                        )
-                    ],
-                    alternate_designations=proposition.subjectVariant.aliases,
+                    gene=[SubmissionVariantGene(symbol=gene_context.name)],
+                    alternate_designations=variant.aliases,
                 )
             ]
         )
@@ -573,6 +575,7 @@ def _get_method_type(
     def _get_submission(
         self,
         statement: GksStatementT,
+        variant: CategoricalVariant | Allele,
         observed_in: list[SubmissionObservedInSomatic],
         variant_hgvs: str | None = None,
         submitted_assembly: Assembly | None = None,
@@ -587,6 +590,7 @@ def _get_submission(
         have record status as `update` rather than `novel`.
 
         :param statement: GKS statement instance to transform
+        :param variant: Variant associated to statement
         :param observed_in: List of distinct ClinVar somatic observations associated
             with the statement
         :param variant_hgvs: The HGVS expression for a variant, if found
@@ -616,17 +620,31 @@ def records_to_submission_container(
         submissions = []
 
         for statement in statements:
+            statement_id = statement.id
             variant = statement.proposition.subjectVariant
 
-            if not isinstance(variant, CategoricalVariant):
+            if isinstance(variant, MolecularVariation):
+                if not isinstance(variant.root, Allele):
+                    logger.warning(
+                        "Skipping statement. Molecular Variation is not an Allele for statement ID: %s",
+                        statement_id,
+                    )
+                    continue
+
+                variant = variant.root
+            elif not isinstance(variant, CategoricalVariant):
+                logger.warning(
+                    "Skipping statement. Variant is not a Categorical Variant or MolecularVariation for statement ID: %s",
+                    statement_id,
+                )
                 continue
 
             variant_hgvs = self._get_variant_hgvs(variant)
 
             if not variant_hgvs:
                 logger.warning(
                     "Skipping statement. No HGVS found for statement ID: %s",
-                    statement.id,
+                    statement_id,
                 )
                 continue
 
@@ -641,19 +659,19 @@ def records_to_submission_container(
             if not observed_in:
                 logger.warning(
                     "Skipping statement. No observed_in found for statement ID: %s",
-                    statement.id,
+                    statement_id,
                 )
                 continue
 
             submissions.append(
                 self._get_submission(
                     statement,
+                    variant,
                     observed_in,
                     variant_hgvs=variant_hgvs,
                     submitted_assembly=batch_metadata.submitted_assembly,
                 )
             )
-
         return SubmissionContainer(
             assertion_criteria=self.assertion_criteria,
             **{self.submission_container_attribute: submissions},

clinvar_this/io/gks_json/clinical_impact_transformer.py

@@ -16,11 +16,12 @@
     PrognosticPredicate,
     TherapeuticResponsePredicate,
 )
+from ga4gh.vrs.models import Allele
+from ga4gh.cat_vrs.models import CategoricalVariant
 
 from clinvar_api.models import (
     Assembly,
     CitationDb,
-    RecordStatus,
     SubmissionAssertionCriteria,
     SubmissionClinicalImpactSubmission,
 )
@@ -84,6 +85,7 @@ class ClinicalImpactTransformer(
     def _get_submission(
         self,
         statement: VariantClinicalSignificanceStatement,
+        variant: CategoricalVariant | Allele,
         observed_in: list[SubmissionObservedInSomatic],
         variant_hgvs: str | None = None,
         submitted_assembly: Assembly | None = None,
@@ -103,6 +105,7 @@ def _get_submission(
         have record status as `update` rather than `novel`.
 
         :param statement: GKS statement (therapeutic, diagnostic, or prognostic) to transform
+        :param variant: Variant associated to statement
         :param observed_in: List of distinct observations
         :param variant_hgvs: The HGVS expression for a variant, if found
         :param submitted_assembly: The genome assembly used to call the variant.
@@ -119,12 +122,14 @@ def _get_submission(
             drug_for_therapeutic_assertion = None
 
         return SubmissionClinicalImpactSubmission(
-            local_id=proposition.subjectVariant.id or proposition.subjectVariant.name,
+            local_id=variant.id or variant.name,
             submitted_assembly=submitted_assembly,
             local_key=statement.id,
             observed_in=observed_in,
             condition_set=self._get_condition_set(proposition),
-            variant_set=self._get_variant_set(proposition, variant_hgvs=variant_hgvs),
+            variant_set=self._get_variant_set(
+                proposition.geneContextQualifier, variant, variant_hgvs=variant_hgvs
+            ),
             clinical_impact_classification=SomaticClinicalImpactClassification(
                 clinical_impact_classification_description=_IMPACT_CLASS_MAPPING[
                     statement.classification.primaryCoding.code.root

clinvar_this/io/gks_json/oncogenicity_transformer.py

@@ -9,11 +9,12 @@
 from clinvar_api.models import (
     Assembly,
     CitationDb,
-    RecordStatus,
     SubmissionAssertionCriteria,
     SubmissionOncogenicitySubmission,
     SomaticOncogenicityClassification,
 )
+from ga4gh.vrs.models import Allele
+from ga4gh.cat_vrs.models import CategoricalVariant
 from ga4gh.va_spec.ccv_2022 import VariantOncogenicityStatement
 from clinvar_api.models.sub_payload import (
     SubmissionObservedInSomatic,
@@ -33,6 +34,7 @@ class OncogenicityTransformer(GksJsonTransformer[VariantOncogenicityStatement]):
     def _get_submission(
         self,
         statement: VariantOncogenicityStatement,
+        variant: CategoricalVariant | Allele,
         observed_in: list[SubmissionObservedInSomatic],
         variant_hgvs: str | None = None,
         submitted_assembly: Assembly | None = None,
@@ -49,6 +51,7 @@ def _get_submission(
         have record status as `update` rather than `novel`.
 
         :param statement: GKS statement (oncogenicity) to transform
+        :param variant: Variant associated to statement
         :param observed_in: List of distinct observations
         :param variant_hgvs: The HGVS expression for a variant, if found
         :param submitted_assembly: The genome assembly used to call the variant.
@@ -59,12 +62,14 @@ def _get_submission(
         proposition = statement.proposition
 
         return SubmissionOncogenicitySubmission(
-            local_id=proposition.subjectVariant.id or proposition.subjectVariant.name,
+            local_id=variant.id or variant.name,
             submitted_assembly=submitted_assembly,
             local_key=statement.id,
             observed_in=observed_in,
             condition_set=self._get_condition_set(proposition),
-            variant_set=self._get_variant_set(proposition, variant_hgvs=variant_hgvs),
+            variant_set=self._get_variant_set(
+                proposition.geneContextQualifier, variant, variant_hgvs=variant_hgvs
+            ),
             oncogenicity_classification=SomaticOncogenicityClassification(
                 oncogenicity_classification_description=statement.classification.primaryCoding.code.root.capitalize(),
                 comment=self._get_comment(statement),

docs/file_formats.rst

@@ -164,9 +164,17 @@ The following information is required or must be derivable from each statement:
 
 - ``proposition.subjectVariant``
 
-  - Expects a GA4GH VA-Spec Categorical Variant.
-  - HGVS expressions are extracted from the first ``DefiningAlleleConstraint`` in ``proposition.subjectVariant.constraints``.
-  - If no supported constraint is present, HGVS expressions may be provided in an ``expressions`` extension on ``proposition.subjectVariant``.
+  - Expects a GA4GH Cat-VRS Categorical Variant or VRS Allele.
+
+    - For Categorical Variants:
+
+      - HGVS expressions are extracted from the ``expressions`` field in the first ``DefiningAlleleConstraint`` in ``proposition.subjectVariant.constraints``.
+      - If no supported constraint is present, HGVS expressions may be provided in an ``expressions`` extension on ``proposition.subjectVariant``.
+
+    - For Alleles:
+
+      - HGVS expressions are extracted from ``proposition.subjectVariant.expressions``.
+
   - RefSeq transcript HGVS expressions are preferred over RefSeq genomic HGVS expressions.
   - The selected HGVS expression is used for the ClinVar variant description.
   - ``proposition.subjectVariant.id`` or ``proposition.subjectVariant.name`` is used as the ClinVar local ID.

tests/clinvar_this/io/gks_json/test_clinical_impact_transformer.py

@@ -1,5 +1,6 @@
 """Module for testing GKS Clinical Impact Transformer"""
 
+from copy import deepcopy
 import re
 
 from deepdiff import DeepDiff
@@ -75,6 +76,38 @@ def civic_aid7(clinical_impact_gks_json_data):
     return VariantClinicalSignificanceStatement(**clinical_impact_gks_json_data[1])
 
 
+@pytest.fixture(scope="module")
+def civic_aid7_allele(clinical_impact_gks_json_data):
+    """Create test fixture for CIViC AID7 where a VRS Allele is used instead"""
+    params = deepcopy(clinical_impact_gks_json_data[1])
+    subject_variant = {
+        "id": "ga4gh:VA.W6xsV-aFm9yT2Bic5cFAV2j0rll6KK5R",
+        "type": "Allele",
+        "name": "NM_004333.6:c.1799T>A",
+        "digest": "W6xsV-aFm9yT2Bic5cFAV2j0rll6KK5R",
+        "expressions": [{"syntax": "hgvs.c", "value": "NM_004333.6:c.1799T>A"}],
+        "location": {
+            "id": "ga4gh:SL.8HBKs9fzlT3tKWlM03REjkg_0Om6Y33U",
+            "type": "SequenceLocation",
+            "digest": "8HBKs9fzlT3tKWlM03REjkg_0Om6Y33U",
+            "sequenceReference": {
+                "type": "SequenceReference",
+                "refgetAccession": "SQ.aKMPEJgmlZXt_F6gRY5cUG3THH2n-GUa",
+                "moleculeType": "RNA",
+            },
+            "start": 2024,
+            "end": 2025,
+        },
+        "state": {"type": "LiteralSequenceExpression", "sequence": "A"},
+    }
+
+    for el in params["hasEvidenceLines"]:
+        el["targetProposition"]["subjectVariant"] = subject_variant
+
+    params["proposition"]["subjectVariant"] = subject_variant
+    return VariantClinicalSignificanceStatement(**params)
+
+
 @pytest.fixture(scope="module")
 def civic_aid20(clinical_impact_gks_json_data):
     """Create test fixture for CIViC AID20"""
@@ -527,3 +560,15 @@ def test_contributions(clinical_impact_transformer, civic_aid200, civic_metadata
         ].clinical_impact_classification.date_last_evaluated
         == "2026-04-16"
     )
+
+
+def test_vrs_allele(clinical_impact_transformer, civic_aid7_allele, civic_metadata):
+    """Test that VRS Alleles work correctly"""
+    actual = clinical_impact_transformer.records_to_submission_container(
+        [civic_aid7_allele], civic_metadata
+    )
+    assert len(actual.clinical_impact_submission) == 1
+    assert len(actual.clinical_impact_submission[0].variant_set.variant) == 1
+    assert actual.clinical_impact_submission[0].variant_set.variant[0].model_dump(
+        exclude_none=True
+    ) == {"hgvs": "NM_004333.6:c.1799T>A", "gene": [{"symbol": "BRAF"}]}