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,11 @@ async def _create_score_calibration(
             db.add(pub)
             db.flush()
 
+    if calibration_create.superseded_calibration_urn:
+        superseded_calibration = validate_superseded_score_calibration(db, calibration_create, user_data)
+    else:
+        superseded_calibration = None
+
     calibration = ScoreCalibration(
         **calibration_create.model_dump(
             by_alias=False,
@@ -185,12 +193,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 +220,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 +261,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 +278,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 +331,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,13 +523,17 @@ 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
     calibration.modified_by = user
 
     db.add(calibration)
+
+    if calibration.superseded_calibration and calibration.superseded_calibration.primary:
+        promote_score_calibration_to_primary(db, calibration, user, force=True)
+
     return calibration
 
 
@@ -552,6 +566,7 @@ def promote_score_calibration_to_primary(
               - If the calibration is already primary.
               - If the calibration is research-use-only.
               - If the calibration is private.
+              - If the calibration is superseded.
               - If another primary calibration exists for the score set and force is False.
 
     Side Effects:
@@ -571,6 +586,9 @@ def promote_score_calibration_to_primary(
     if calibration.private:
         raise ValueError("Cannot promote a private calibration to primary.")
 
+    if calibration.superseding_calibration:
+        raise ValueError("Cannot promote a superseded calibration to primary.")
+
     existing_primary_calibrations = (
         db.query(ScoreCalibration)
         .filter(
@@ -652,6 +670,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,13 +132,25 @@ 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()
     )
 
-    permitted_calibrations = [
+    visible_calibrations = [
         calibration for calibration in calibrations if has_permission(user_data, calibration, Action.READ).permitted
     ]
+
+    superseded_ids = [sc.superseded_calibration_id for sc in visible_calibrations if
+                      sc.superseded_calibration_id is not None]
+
+    permitted_calibrations = [sc for sc in visible_calibrations if sc.id not in superseded_ids]
+
+    # Solve Pydantic model validation error
+    for sc in permitted_calibrations:
+        sc.superseded_calibration = None
+        sc.superseding_calibration = None
+
     if not permitted_calibrations:
         logger.debug("No score calibrations found for the requested score set", extra=logging_context())
         raise HTTPException(status_code=404, detail="No score calibrations found for the requested score set")
@@ -338,9 +350,12 @@ async def create_score_calibration_route(
                 detail=[{"loc": [e.custom_loc or "classesFile"], "msg": str(e), "type": "value_error"}],
             )
 
-    created_calibration = await create_score_calibration_in_score_set(
-        db, calibration, user_data.user, variant_classes if classes_file else None
-    )
+    try:
+        created_calibration = await create_score_calibration_in_score_set(
+            db, calibration, user_data, variant_classes if classes_file else None
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=422, detail=str(e))
 
     db.commit()
     db.refresh(created_calibration)
@@ -598,6 +613,10 @@ async def promote_score_calibration_to_primary_route(
         logger.debug("Private score calibrations cannot be promoted to primary", extra=logging_context())
         raise HTTPException(status_code=400, detail="Private score calibrations cannot be promoted to primary")
 
+    if item.superseding_calibration:
+        logger.debug("Superseded score calibrations cannot be promoted to primary", extra=logging_context())
+        raise HTTPException(status_code=400, detail="Superseded score calibrations cannot be promoted to primary")
+
     # We've already checked whether the item matching the calibration URN is primary, so this
     # will necessarily be a different calibration, if it exists.
     existing_primary_calibration = next((c for c in item.score_set.score_calibrations if c.primary), None)

src/mavedb/routers/score_sets.py

@@ -604,7 +604,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
 
@@ -1703,7 +1714,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