Add superseded_calibration in Calibration. Modified related functions and tests.

alembic/versions/adb481b7c60b_add_calibration_superseded_column.py

@@ -0,0 +1,31 @@
+"""add_calibration_superseded_column
+
+Revision ID: adb481b7c60b
+Revises: 398067c53257
+Create Date: 2026-06-01 16:45:35.507837
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = 'adb481b7c60b'
+down_revision = '398067c53257'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('score_calibrations', sa.Column('replaces_id', sa.Integer(), nullable=True))
+    op.create_index(op.f('ix_score_calibrations_replaces_id'), 'score_calibrations', ['replaces_id'], unique=False)
+    op.create_foreign_key(None, 'score_calibrations', 'score_calibrations', ['replaces_id'], ['id'])
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'score_calibrations', type_='foreignkey')
+    op.drop_index(op.f('ix_score_calibrations_replaces_id'), table_name='score_calibrations')
+    op.drop_column('score_calibrations', 'replaces_id')
+    # ### end Alembic commands ###

src/mavedb/lib/score_calibrations.py

@@ -9,6 +9,8 @@
 
 from mavedb.lib.acmg import find_or_create_acmg_classification
 from mavedb.lib.identifiers import find_or_create_publication_identifier
+from mavedb.lib.permissions import Action, has_permission
+from mavedb.lib.types.authentication import UserData
 from mavedb.lib.types.score_calibrations import ClassificationDict
 from mavedb.lib.validation.constants.general import (
     calibration_class_column_name,
@@ -25,6 +27,7 @@
 from mavedb.models.user import User
 from mavedb.models.variant import Variant
 from mavedb.view_models import score_calibration
+from mavedb.view_models.score_calibration import ScoreCalibrationCreate
 
 
 def create_functional_classification(
@@ -97,7 +100,7 @@ def create_functional_classification(
 async def _create_score_calibration(
     db: Session,
     calibration_create: score_calibration.ScoreCalibrationCreate,
-    user: User,
+    user_data: UserData,
     variant_classes: Optional[ClassificationDict] = None,
     containing_score_set: Optional[ScoreSet] = None,
 ) -> ScoreCalibration:
@@ -176,6 +179,8 @@ async def _create_score_calibration(
             db.add(pub)
             db.flush()
 
+    superseded_calibration = validate_superseded_score_calibration(db, calibration_create, user_data)
+
     calibration = ScoreCalibration(
         **calibration_create.model_dump(
             by_alias=False,
@@ -185,12 +190,14 @@ async def _create_score_calibration(
                 "evidence_sources",
                 "method_sources",
                 "score_set_urn",
+                "superseded_calibration_urn"
             },
         ),
         publication_identifier_associations=calibration_pub_assocs,
         functional_classifications=[],
-        created_by=user,
-        modified_by=user,
+        superseded_calibration=superseded_calibration,
+        created_by=user_data.user,
+        modified_by=user_data.user,
     )  # type: ignore[call-arg]
 
     if containing_score_set:
@@ -210,7 +217,7 @@ async def _create_score_calibration(
 async def create_score_calibration_in_score_set(
     db: Session,
     calibration_create: score_calibration.ScoreCalibrationCreate,
-    user: User,
+    user_data: UserData,
     variant_classes: Optional[ClassificationDict] = None,
 ) -> ScoreCalibration:
     """
@@ -251,9 +258,9 @@ async def create_score_calibration_in_score_set(
         raise ValueError("score_set_urn must be provided to create a score calibration within a score set.")
 
     containing_score_set = db.query(ScoreSet).where(ScoreSet.urn == calibration_create.score_set_urn).one()
-    calibration = await _create_score_calibration(db, calibration_create, user, variant_classes, containing_score_set)
+    calibration = await _create_score_calibration(db, calibration_create, user_data, variant_classes, containing_score_set)
 
-    if user.username in [contributor.orcid_id for contributor in containing_score_set.contributors] + [
+    if user_data.user.username in [contributor.orcid_id for contributor in containing_score_set.contributors] + [
         containing_score_set.created_by.username,
         containing_score_set.modified_by.username,
     ]:
@@ -268,7 +275,7 @@ async def create_score_calibration_in_score_set(
 async def create_score_calibration(
     db: Session,
     calibration_create: score_calibration.ScoreCalibrationCreate,
-    user: User,
+    user_data: UserData,
     variant_classes: Optional[ClassificationDict] = None,
 ) -> ScoreCalibration:
     """
@@ -321,7 +328,7 @@ async def create_score_calibration(
         raise ValueError("score_set_urn must not be provided to create a score calibration outside a score set.")
 
     created_calibration = await _create_score_calibration(
-        db, calibration_create, user, variant_classes, containing_score_set=None
+        db, calibration_create, user_data, variant_classes, containing_score_set=None
     )
 
     db.add(created_calibration)
@@ -513,7 +520,7 @@ def publish_score_calibration(db: Session, calibration: ScoreCalibration, user:
     This function adds the modified calibration to the session but does not commit;
     the caller is responsible for committing the transaction.
     """
-    if not calibration.private:
+    if calibration.private is False: # It's possible None value
         raise ValueError("Calibration is already published.")
 
     calibration.private = False
@@ -652,6 +659,35 @@ def delete_score_calibration(db: Session, calibration: ScoreCalibration) -> None
     return None
 
 
+def validate_superseded_score_calibration(
+    db: Session,
+    calibration_create: ScoreCalibrationCreate,
+    user_data: Optional[UserData],
+) -> Optional[ScoreCalibration]:
+    if not calibration_create.superseded_calibration_urn:
+        return None
+
+    superseded_calibration = (
+        db.query(ScoreCalibration)
+        .where(ScoreCalibration.urn == calibration_create.superseded_calibration_urn)
+        .one_or_none()
+    )
+
+    if superseded_calibration is None:
+        raise ValueError("Superseded calibration does not exist.")
+
+    if not has_permission(user_data, superseded_calibration, Action.READ).permitted:
+        raise ValueError("No access right to supersede this calibration.")
+
+    if superseded_calibration.private:
+        raise ValueError("Cannot supersede a private calibration. Please edit it instead.")
+
+    if superseded_calibration.superseding_calibration:
+        raise ValueError("Cannot supersede a superseded calibration. Please edit it instead.")
+
+    return superseded_calibration
+
+
 def variants_for_functional_classification(
     db: Session,
     functional_classification: ScoreCalibrationFunctionalClassification,

src/mavedb/lib/validation/urn_re.py

@@ -32,6 +32,14 @@
 MAVEDB_COLLECTION_URN_PATTERN = r"urn:mavedb:collection-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
 MAVEDB_COLLECTION_URN_RE = re.compile(MAVEDB_COLLECTION_URN_PATTERN)
 
+# Temp calibration URN
+MAVEDB_TMP_CALIBRATION_URN_PATTERN = r"tmp:mavedb.calibration-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
+MAVEDB_TMP_CALIBRATION_URN_RE = re.compile(MAVEDB_TMP_CALIBRATION_URN_PATTERN)
+
+# Calibration URN
+MAVEDB_CALIBRATION_URN_PATTERN = r"urn:mavedb:calibration-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
+MAVEDB_CALIBRATION_URN_RE = re.compile(MAVEDB_CALIBRATION_URN_PATTERN)
+
 # Any URN
 MAVEDB_ANY_URN_PATTERN = "|".join(
     [

src/mavedb/models/score_calibration.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from datetime import date
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 from sqlalchemy import Boolean, Column, Date, Float, ForeignKey, Integer, String
 from sqlalchemy.dialects.postgresql import JSONB
@@ -60,6 +60,17 @@ class ScoreCalibration(Base):
 
     calibration_metadata = Column(JSONB(none_as_null=True), nullable=True)
 
+    superseded_calibration_id = Column("replaces_id", Integer, ForeignKey("score_calibrations.id"), index=True, nullable=True)
+    superseded_calibration: Mapped[Optional["ScoreCalibration"]] = relationship(
+        "ScoreCalibration",
+        uselist=False,
+        foreign_keys="ScoreCalibration.superseded_calibration_id",
+        remote_side=[id],
+    )
+    superseding_calibration: Mapped[Optional["ScoreCalibration"]] = relationship(
+        "ScoreCalibration", uselist=False, back_populates="superseded_calibration"
+    )
+
     created_by_id = Column(Integer, ForeignKey("users.id"), index=True, nullable=False)
     created_by: Mapped["User"] = relationship("User", foreign_keys="ScoreCalibration.created_by_id")
     modified_by_id = Column(Integer, ForeignKey("users.id"), index=True, nullable=False)

src/mavedb/routers/score_calibrations.py

@@ -132,6 +132,7 @@ async def get_score_calibrations_for_score_set(
     calibrations = (
         db.query(ScoreCalibration)
         .filter(ScoreCalibration.score_set_id == score_set.id)
+        .filter(~ScoreCalibration.superseding_calibration.has(ScoreCalibration.private.is_(False)))
         .options(selectinload(ScoreCalibration.score_set).selectinload(ScoreSet.contributors))
         .all()
     )
@@ -339,7 +340,7 @@ async def create_score_calibration_route(
             )
 
     created_calibration = await create_score_calibration_in_score_set(
-        db, calibration, user_data.user, variant_classes if classes_file else None
+        db, calibration, user_data, variant_classes if classes_file else None
     )
 
     db.commit()

src/mavedb/routers/score_sets.py

@@ -602,7 +602,18 @@ async def fetch_score_set_by_urn(
     if item.superseding_score_set and not has_permission(user, item.superseding_score_set, Action.READ).permitted:
         item.superseding_score_set = None
 
-    item.score_calibrations = [sc for sc in item.score_calibrations if has_permission(user, sc, Action.READ).permitted]
+    visible_calibrations = [sc for sc in item.score_calibrations if has_permission(user, sc, Action.READ).permitted]
+
+    superseded_ids = [sc.superseded_calibration_id for sc in visible_calibrations if sc.superseded_calibration_id is not None]
+
+    available_calibrations = [sc for sc in visible_calibrations if sc.id not in superseded_ids]
+
+    # Solve Pydantic model validation error
+    for sc in available_calibrations:
+        sc.superseded_calibration = None
+        sc.superseding_calibration = None
+
+    item.score_calibrations = available_calibrations
 
     return item
 
@@ -1742,7 +1753,7 @@ async def create_score_set(
                 )
 
             created_calibration_item = await create_score_calibration(
-                db, calibration_create, user_data.user, variant_classes=None
+                db, calibration_create, user_data, variant_classes=None
             )
             created_calibration_item.investigator_provided = True  # necessarily true on score set creation
             score_calibrations.append(created_calibration_item)

src/mavedb/scripts/load_calibration_csv.py

@@ -100,6 +100,7 @@
 from mavedb.lib.acmg import ACMGCriterion, StrengthOfEvidenceProvided
 from mavedb.lib.oddspaths import oddspaths_evidence_strength_equivalent
 from mavedb.lib.score_calibrations import create_score_calibration_in_score_set
+from mavedb.lib.types.authentication import UserData
 from mavedb.models import score_calibration
 from mavedb.models.enums.functional_classification import FunctionalClassification as FunctionalClassifcationOptions
 from mavedb.models.score_set import ScoreSet
@@ -414,8 +415,9 @@ def main(db: Session, csv_path: str, delimiter: str, overwrite: bool, purge_publ
 
                 system_user = db.query(User).filter(User.id == 1).one()
                 calibration_user = score_set.created_by if calibration_is_investigator_provided else system_user
+                calibration_user_data = UserData(calibration_user, calibration_user.roles)
                 new_calibration_object = asyncio.run(
-                    create_score_calibration_in_score_set(db, created_score_calibration, calibration_user)
+                    create_score_calibration_in_score_set(db, created_score_calibration, calibration_user_data)
                 )
                 new_calibration_object.primary = primary
                 new_calibration_object.private = False

src/mavedb/scripts/load_excalibr_calibrations.py

@@ -49,6 +49,7 @@
 from sqlalchemy.orm import Session
 
 from mavedb.lib.score_calibrations import create_score_calibration_in_score_set
+from mavedb.lib.types.authentication import UserData
 from mavedb.models.enums.functional_classification import FunctionalClassification as FunctionalClassificationOptions
 from mavedb.models.score_calibration import ScoreCalibration
 from mavedb.models.score_set import ScoreSet
@@ -238,8 +239,9 @@ def main(db: Session, csv_path: str, dataset_map: str, overwrite: bool, remove:
                     method_sources=[EXCALIBR_CALIBRATION_CITATION],
                 )
 
+                system_user_data = UserData(system_user, system_user.roles)
                 new_calibration_object = asyncio.run(
-                    create_score_calibration_in_score_set(db, score_calibration_create, system_user)
+                    create_score_calibration_in_score_set(db, score_calibration_create, system_user_data)
                 )
                 new_calibration_object.primary = False
                 new_calibration_object.private = False

src/mavedb/view_models/score_calibration.py

@@ -10,6 +10,7 @@
 from pydantic import Field, field_validator, model_validator
 
 from mavedb.lib.oddspaths import oddspaths_evidence_strength_equivalent
+from mavedb.lib.validation import urn_re
 from mavedb.lib.validation.exceptions import ValidationError
 from mavedb.lib.validation.transform import (
     transform_score_calibration_publication_identifiers,
@@ -477,11 +478,24 @@ def functional_classifications_require_publication_sources(
 class ScoreCalibrationCreate(ScoreCalibrationModify):
     """Model used to create a new score calibration."""
 
+    superseded_calibration_urn: Optional[str] = None
     functional_classifications: Optional[Sequence[FunctionalClassificationCreate]] = None
     threshold_sources: Sequence[PublicationIdentifierCreate]
     evidence_sources: Sequence[PublicationIdentifierCreate]
     method_sources: Sequence[PublicationIdentifierCreate]
 
+    @field_validator("superseded_calibration_urn")
+    def validate_superseded_calibration_urn(cls, v: Optional[str]) -> Optional[str]:
+        if not v:
+            return None
+
+        if urn_re.MAVEDB_CALIBRATION_URN_RE.fullmatch(v) is None:
+            if urn_re.MAVEDB_TMP_CALIBRATION_URN_RE.fullmatch(v) is None:
+                raise ValueError(f"'{v}' is not a valid calibration URN")
+            else:
+                raise ValueError("cannot supersede a private calibration - please edit it instead")
+
+        return v
 
 class SavedScoreCalibration(ScoreCalibrationBase):
     """Persisted score calibration model (includes identifiers and source lists)."""
@@ -501,6 +515,8 @@ class SavedScoreCalibration(ScoreCalibrationBase):
     threshold_sources: Sequence[SavedPublicationIdentifier]
     evidence_sources: Sequence[SavedPublicationIdentifier]
     method_sources: Sequence[SavedPublicationIdentifier]
+    superseded_calibration: Optional[ScoreCalibrationBase] = None
+    superseding_calibration: Optional[ScoreCalibrationBase] = None
 
     created_by: Optional[SavedUser] = None
     modified_by: Optional[SavedUser] = None

tests/helpers/util/score_calibration.py

@@ -3,8 +3,8 @@
 import jsonschema
 
 from mavedb.lib.score_calibrations import create_score_calibration_in_score_set
+from mavedb.lib.types.authentication import UserData
 from mavedb.models.score_calibration import ScoreCalibration
-from mavedb.models.user import User
 from mavedb.view_models.score_calibration import ScoreCalibrationCreate, ScoreCalibrationWithScoreSetUrn
 from tests.helpers.constants import TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED
 
@@ -14,12 +14,12 @@
 
 
 async def create_test_range_based_score_calibration_in_score_set(
-    db: "Session", score_set_urn: str, user: User
+    db: "Session", score_set_urn: str, user_data: UserData
 ) -> ScoreCalibration:
     calibration_create = ScoreCalibrationCreate(
         **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, score_set_urn=score_set_urn
     )
-    created_score_calibration = await create_score_calibration_in_score_set(db, calibration_create, user)
+    created_score_calibration = await create_score_calibration_in_score_set(db, calibration_create, user_data)
     assert created_score_calibration is not None
 
     db.commit()