Merge pull request #439 from VariantEffect/feature/bencap/438/add-odds-path-to-score-range-data-model

Add OddsPath Property to ScoreRange Data Model

Author: bencap

src/mavedb/view_models/odds_path.py

@@ -0,0 +1,44 @@
+from typing import Literal, Optional
+from pydantic import validator
+
+from mavedb.view_models import record_type_validator, set_record_type
+from mavedb.view_models.base.base import BaseModel
+
+
+class OddsPathBase(BaseModel):
+    ratio: float
+    evidence: Optional[
+        Literal[
+            "BS3_STRONG",
+            "BS3_MODERATE",
+            "BS3_SUPPORTING",
+            "INDETERMINATE",
+            "PS3_VERY_STRONG",
+            "PS3_STRONG",
+            "PS3_MODERATE",
+            "PS3_SUPPORTING",
+        ]
+    ] = None
+
+
+class OddsPathModify(OddsPathBase):
+    @validator("ratio")
+    def ratio_must_be_positive(cls, value: float) -> float:
+        if value < 0:
+            raise ValueError("OddsPath value must be greater than or equal to 0")
+
+        return value
+
+
+class OddsPathCreate(OddsPathModify):
+    pass
+
+
+class SavedOddsPath(OddsPathBase):
+    record_type: str = None  # type: ignore
+
+    _record_type_factory = record_type_validator()(set_record_type)
+
+
+class OddsPath(SavedOddsPath):
+    pass

src/mavedb/view_models/score_range.py

@@ -0,0 +1,88 @@
+from typing import Optional, Literal, Any, Sequence
+from pydantic import validator
+
+from mavedb.lib.validation.exceptions import ValidationError
+from mavedb.lib.validation.utilities import inf_or_float
+from mavedb.view_models import record_type_validator, set_record_type
+from mavedb.view_models.base.base import BaseModel
+from mavedb.view_models.publication_identifier import PublicationIdentifierBase
+from mavedb.view_models.odds_path import OddsPathCreate, OddsPathBase, OddsPathModify, SavedOddsPath, OddsPath
+
+
+### Range model
+
+
+class ScoreRangeBase(BaseModel):
+    label: str
+    description: Optional[str]
+    classification: Literal["normal", "abnormal", "not_specified"]
+    # Purposefully vague type hint because of some odd JSON Schema generation behavior.
+    # Typing this as tuple[Union[float, None], Union[float, None]] will generate an invalid
+    # jsonschema, and fail all tests that access the schema. This may be fixed in pydantic v2,
+    # but it's unclear. Even just typing it as Tuple[Any, Any] will generate an invalid schema!
+    range: list[Any]  # really: tuple[Union[float, None], Union[float, None]]
+    odds_path: Optional[OddsPathBase] = None
+
+
+class ScoreRangeModify(ScoreRangeBase):
+    odds_path: Optional[OddsPathModify] = None
+
+    @validator("range")
+    def ranges_are_not_backwards(cls, field_value: tuple[Any]):
+        if len(field_value) != 2:
+            raise ValidationError("Only a lower and upper bound are allowed.")
+
+        field_value[0] = inf_or_float(field_value[0], True) if field_value[0] is not None else None
+        field_value[1] = inf_or_float(field_value[1], False) if field_value[1] is not None else None
+
+        if inf_or_float(field_value[0], True) > inf_or_float(field_value[1], False):
+            raise ValidationError("The lower bound of the score range may not be larger than the upper bound.")
+        elif inf_or_float(field_value[0], True) == inf_or_float(field_value[1], False):
+            raise ValidationError("The lower and upper bound of the score range may not be the same.")
+
+        return field_value
+
+
+class ScoreRangeCreate(ScoreRangeModify):
+    odds_path: Optional[OddsPathCreate] = None
+
+
+class SavedScoreRange(ScoreRangeBase):
+    record_type: str = None  # type: ignore
+
+    odds_path: Optional[SavedOddsPath] = None
+
+    _record_type_factory = record_type_validator()(set_record_type)
+
+
+class ScoreRange(SavedScoreRange):
+    odds_path: Optional[OddsPath] = None
+
+
+### Ranges wrapper
+
+
+class ScoreSetRangesBase(BaseModel):
+    wt_score: Optional[float] = None
+    ranges: Sequence[ScoreRangeBase]
+    odds_path_source: Optional[Sequence[PublicationIdentifierBase]] = None
+
+
+class ScoreSetRangesModify(ScoreSetRangesBase):
+    ranges: Sequence[ScoreRangeModify]
+
+
+class ScoreSetRangesCreate(ScoreSetRangesModify):
+    ranges: Sequence[ScoreRangeCreate]
+
+
+class SavedScoreSetRanges(ScoreSetRangesBase):
+    record_type: str = None  # type: ignore
+
+    ranges: Sequence[SavedScoreRange]
+
+    _record_type_factory = record_type_validator()(set_record_type)
+
+
+class ScoreSetRanges(SavedScoreSetRanges):
+    ranges: Sequence[ScoreRange]

src/mavedb/view_models/score_set.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from datetime import date
-from typing import Any, Collection, Dict, Optional, Sequence, Literal
+from typing import Any, Collection, Dict, Optional, Sequence
 
 from humps import camelize
 from pydantic import root_validator
@@ -27,6 +27,7 @@
     PublicationIdentifierCreate,
     SavedPublicationIdentifier,
 )
+from mavedb.view_models.score_range import SavedScoreSetRanges, ScoreSetRangesCreate, ScoreSetRanges
 from mavedb.view_models.target_gene import (
     SavedTargetGene,
     ShortTargetGene,
@@ -50,37 +51,6 @@ class Config:
         arbitrary_types_allowed = True
 
 
-class ScoreRange(BaseModel):
-    label: str
-    description: Optional[str]
-    classification: Literal["normal", "abnormal", "not_specified"]
-    # Purposefully vague type hint because of some odd JSON Schema generation behavior.
-    # Typing this as tuple[Union[float, None], Union[float, None]] will generate an invalid
-    # jsonschema, and fail all tests that access the schema. This may be fixed in pydantic v2,
-    # but it's unclear. Even just typing it as Tuple[Any, Any] will generate an invalid schema!
-    range: list[Any]  # really: tuple[Union[float, None], Union[float, None]]
-
-    @validator("range")
-    def ranges_are_not_backwards(cls, field_value: tuple[Any]):
-        if len(field_value) != 2:
-            raise ValidationError("Only a lower and upper bound are allowed.")
-
-        field_value[0] = inf_or_float(field_value[0], True) if field_value[0] is not None else None
-        field_value[1] = inf_or_float(field_value[1], False) if field_value[1] is not None else None
-
-        if inf_or_float(field_value[0], True) > inf_or_float(field_value[1], False):
-            raise ValidationError("The lower bound of the score range may not be larger than the upper bound.")
-        elif inf_or_float(field_value[0], True) == inf_or_float(field_value[1], False):
-            raise ValidationError("The lower and upper bound of the score range may not be the same.")
-
-        return field_value
-
-
-class ScoreRanges(BaseModel):
-    wt_score: Optional[float]
-    ranges: list[ScoreRange]  # type: ignore
-
-
 class ScoreSetGetter(PublicationIdentifiersGetter):
     def get(self, key: Any, default: Any = ...) -> Any:
         if key == "meta_analyzes_score_set_urns":
@@ -110,7 +80,7 @@ class ScoreSetModify(ScoreSetBase):
     secondary_publication_identifiers: Optional[list[PublicationIdentifierCreate]]
     doi_identifiers: Optional[list[DoiIdentifierCreate]]
     target_genes: list[TargetGeneCreate]
-    score_ranges: Optional[ScoreRanges]
+    score_ranges: Optional[ScoreSetRangesCreate]
 
     @validator("title", "short_description", "abstract_text", "method_text")
     def validate_field_is_non_empty(cls, v):
@@ -198,7 +168,7 @@ def target_accession_base_editor_targets_are_consistent(cls, field_value, values
         return field_value
 
     @validator("score_ranges")
-    def score_range_labels_must_be_unique(cls, field_value: Optional[ScoreRanges]):
+    def score_range_labels_must_be_unique(cls, field_value: Optional[ScoreSetRangesCreate]):
         if field_value is None:
             return None
 
@@ -217,7 +187,9 @@ def score_range_labels_must_be_unique(cls, field_value: Optional[ScoreRanges]):
         return field_value
 
     @validator("score_ranges")
-    def score_range_normal_classification_exists_if_wild_type_score_provided(cls, field_value: Optional[ScoreRanges]):
+    def score_range_normal_classification_exists_if_wild_type_score_provided(
+        cls, field_value: Optional[ScoreSetRangesCreate]
+    ):
         if field_value is None:
             return None
 
@@ -231,7 +203,7 @@ def score_range_normal_classification_exists_if_wild_type_score_provided(cls, fi
         return field_value
 
     @validator("score_ranges")
-    def ranges_do_not_overlap(cls, field_value: Optional[ScoreRanges]):
+    def ranges_do_not_overlap(cls, field_value: Optional[ScoreSetRangesCreate]):
         def test_overlap(tp1, tp2) -> bool:
             # Always check the tuple with the lowest lower bound. If we do not check
             # overlaps in this manner, checking the overlap of (0,1) and (1,2) will
@@ -264,7 +236,7 @@ def test_overlap(tp1, tp2) -> bool:
         return field_value
 
     @validator("score_ranges")
-    def wild_type_score_in_normal_range(cls, field_value: Optional[ScoreRanges]):
+    def wild_type_score_in_normal_range(cls, field_value: Optional[ScoreSetRangesCreate]):
         if field_value is None:
             return None
 
@@ -292,6 +264,35 @@ def wild_type_score_in_normal_range(cls, field_value: Optional[ScoreRanges]):
             custom_loc=["body", "scoreRanges", "wtScore"],
         )
 
+    @root_validator()
+    def validate_score_range_odds_path_source_in_publication_identifiers(cls, values):
+        score_ranges: Optional[ScoreSetRangesCreate] = values.get("score_ranges")
+        if score_ranges is None or score_ranges.odds_path_source is None:
+            return values
+
+        if not score_ranges.odds_path_source:
+            return values
+
+        for idx, pub in enumerate(score_ranges.odds_path_source):
+            primary_publication_identifiers = (
+                values.get("primary_publication_identifiers", [])
+                if values.get("primary_publication_identifiers")
+                else []
+            )
+            secondary_publication_identifiers = (
+                values.get("secondary_publication_identifiers", [])
+                if values.get("secondary_publication_identifiers")
+                else []
+            )
+            if pub not in [*primary_publication_identifiers, *secondary_publication_identifiers]:
+                raise ValidationError(
+                    f"Odds path source publication identifier at index {idx} is not defined in score set publications. "
+                    "To use a publication identifier in the odds path source, it must be defined in the primary or secondary publication identifiers.",
+                    custom_loc=["body", "scoreRanges", "oddsPath", "source", idx],
+                )
+
+        return values
+
 
 class ScoreSetCreate(ScoreSetModify):
     """View model for creating a new score set."""
@@ -414,7 +415,7 @@ class SavedScoreSet(ScoreSetBase):
     dataset_columns: Dict
     external_links: Dict[str, ExternalLink]
     contributors: list[Contributor]
-    score_ranges: Optional[ScoreRanges]
+    score_ranges: Optional[SavedScoreSetRanges]
     score_calibrations: Optional[dict[str, Calibration]]
 
     _record_type_factory = record_type_validator()(set_record_type)
@@ -452,6 +453,7 @@ class ScoreSet(SavedScoreSet):
     processing_errors: Optional[dict]
     mapping_state: Optional[MappingState]
     mapping_errors: Optional[dict]
+    score_ranges: Optional[ScoreSetRanges]
 
 
 class ScoreSetWithVariants(ScoreSet):
@@ -484,6 +486,7 @@ class ScoreSetPublicDump(SavedScoreSet):
     processing_errors: Optional[Dict]
     mapping_state: Optional[MappingState]
     mapping_errors: Optional[Dict]
+    score_ranges: Optional[ScoreSetRanges]
 
 
 # ruff: noqa: E402