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 = 'a7f3c2e9b104'
+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_sets.py

@@ -25,6 +25,7 @@
 )
 from mavedb.lib.mave.utils import is_csv_null
 from mavedb.lib.permissions import Action, has_permission
+from mavedb.lib.score_calibrations import find_superseded_score_calibration_tail
 from mavedb.lib.types.authentication import UserData
 from mavedb.lib.validation.constants.general import null_values_list
 from mavedb.lib.validation.utilities import is_null as validate_is_null
@@ -57,6 +58,7 @@
 from mavedb.models.uniprot_offset import UniprotOffset
 from mavedb.models.user import User
 from mavedb.models.variant import Variant
+from mavedb.view_models import score_set
 from mavedb.view_models.search import ScoreSetsSearch, ControlledKeywordFilterOption
 
 if TYPE_CHECKING:
@@ -321,6 +323,33 @@ def score_set_search_filter_options_from_counter(counter: Counter):
     return [{"value": value, "count": count} for value, count in counter.items()]
 
 
+def enrich_score_set_with_num_score_calibrations(
+    item_update: ScoreSet, user_data: Optional[UserData]
+) -> score_set.ScoreSet:
+    """
+    Validate and update the number of score calibration in score set. The superseded score calibration is excluded.
+    Data structure: score_set{score_calibration_urns, num_score_calibrations}
+    """
+    filter_superseded_score_calibration_tails = [
+        find_superseded_score_calibration_tail(score_calibration, Action.READ, user_data) for score_calibration in item_update.score_calibrations
+    ]
+    filtered_score_calibration_urns = sorted(
+        {
+            score_calibration.urn
+            for score_calibration in filter_superseded_score_calibration_tails
+            if score_calibration is not None and score_calibration.urn is not None
+        }
+    )
+
+    updated_score_set = score_set.ScoreSet.model_validate(item_update).copy(
+        update={
+            "num_score_calibrations": len(filtered_score_calibration_urns),
+            "score_calibration_urns": filtered_score_calibration_urns,
+        }
+    )
+    return updated_score_set
+
+
 def fetch_score_set_search_filter_options(
     db: Session, requester: Optional[UserData], owner_or_contributor: Optional[User], search: ScoreSetsSearch
 ):
@@ -351,13 +380,13 @@ def fetch_score_set_search_filter_options(
     #   - Use parallelization (e.g., multiprocessing or concurrent.futures) for large datasets
     #   - Pre-fetch or denormalize target/publication data in the DB query
     #   - Profile and refactor nested attribute lookups to minimize Python overhead
-    for score_set in score_sets:
+    for ss in score_sets:
         # Check read permission for each score set, skip if no permission
-        if not has_permission(requester, score_set, Action.READ).permitted:
+        if not has_permission(requester, ss, Action.READ).permitted:
             continue
 
         # Target related options
-        for target in getattr(score_set, "target_genes", []):
+        for target in getattr(ss, "target_genes", []):
             category = getattr(target, "category", None)
             if category:
                 target_category_counter[category] += 1
@@ -380,7 +409,7 @@ def fetch_score_set_search_filter_options(
                 target_accession_counter[accession] += 1
 
         # Publication related options
-        for publication_association in getattr(score_set, "publication_identifier_associations", []):
+        for publication_association in getattr(ss, "publication_identifier_associations", []):
             publication = getattr(publication_association, "publication", None)
 
             authors = getattr(publication, "authors", [])
@@ -398,7 +427,7 @@ def fetch_score_set_search_filter_options(
                 publication_journal_counter[journal] += 1
 
         # Controlled keywords related options
-        for controlled_keyword in getattr(score_set.experiment, "keyword_objs", []):
+        for controlled_keyword in getattr(ss.experiment, "keyword_objs", []):
             keyword = getattr(controlled_keyword, "controlled_keyword", [])
             if not keyword:
                 continue

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

@@ -1,5 +1,5 @@
 import logging
-from typing import Optional
+from typing import Any, Optional
 
 from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
 from sqlalchemy.orm import Session, selectinload
@@ -21,9 +21,10 @@
     modify_score_calibration,
     promote_score_calibration_to_primary,
     publish_score_calibration,
+    search_score_calibrations as _search_score_calibrations,
     variant_classification_df_to_dict,
 )
-from mavedb.lib.score_sets import csv_data_to_df
+from mavedb.lib.score_sets import csv_data_to_df, enrich_score_set_with_num_score_calibrations
 from mavedb.lib.types.authentication import UserData
 from mavedb.lib.validation.constants.general import calibration_class_column_name, calibration_variant_column_name
 from mavedb.lib.validation.dataframe.calibration import validate_and_standardize_calibration_classes_dataframe
@@ -33,6 +34,7 @@
 from mavedb.models.score_set import ScoreSet
 from mavedb.routers.shared import ACCESS_CONTROL_ERROR_RESPONSES, PUBLIC_ERROR_RESPONSES
 from mavedb.view_models import score_calibration
+from mavedb.view_models.search import ScoreCalibrationsSearch, ScoreCalibrationsSearchResponse
 
 logger = logging.getLogger(__name__)
 
@@ -132,13 +134,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 +352,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 +615,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)
@@ -708,6 +729,31 @@ def publish_score_calibration_route(
     return item
 
 
+@router.post(
+    "/me/search",
+    status_code=200,
+    summary="Search my calibrations",
+    responses={**ACCESS_CONTROL_ERROR_RESPONSES},
+    response_model=ScoreCalibrationsSearchResponse,
+)
+def search_my_score_calibrations(
+    search: ScoreCalibrationsSearch,
+    db: Session = Depends(deps.get_db),
+    user_data: UserData = Depends(require_current_user),
+) -> Any:
+    """
+    Search calibrations created by the current user.
+    """
+    score_calibrations, num_score_calibrations = _search_score_calibrations(db, user_data.user, search).values()
+    enriched_score_calibrations = []
+    for sc in score_calibrations:
+        enriched_score_calibration = enrich_score_set_with_num_score_calibrations(sc.score_set, user_data)
+        response_item = score_calibration.ScoreCalibration.model_validate(sc).copy(update={"score_calibration": enriched_score_calibration})
+        enriched_score_calibrations.append(response_item)
+
+    return {"score_calibrations": enriched_score_calibrations, "num_score_calibrations": num_score_calibrations}
+
+
 @router.get(
     "/{urn}/functional-classifications/{classification_id}/variants",
     response_model=score_calibration.FunctionalClassificationVariants,

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