Change to a dual unique constraint on ID and tag for data collection groups

Author: d-j-hatton

src/murfey/cli/inject_spa_processing.py

@@ -132,6 +132,7 @@ def run():
             .where(DataCollectionGroup.session_id == args.session_id)
             .where(DataCollectionGroup.tag == args.tag)
             .where(DataCollection.dcg_id == DataCollectionGroup.id)
+            .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
             .where(ProcessingJob.dc_id == DataCollection.id)
             .where(AutoProcProgram.pj_id == ProcessingJob.id)
             .where(ProcessingJob.recipe == "em-spa-preprocess")

src/murfey/cli/spa_ispyb_messages.py

@@ -336,6 +336,7 @@ def run():
         .where(db.DataCollectionGroup.session_id == args.session_id)
         .where(db.DataCollectionGroup.tag == args.tag)
         .where(db.DataCollection.dcg_id == db.DataCollectionGroup.id)
+        .where(db.DataCollection.dcg_tag == db.DataCollectionGroup.tag)
         .where(db.ProcessingJob.dc_id == db.DataCollection.id)
         .where(db.AutoProcProgram.pj_id == db.ProcessingJob.id)
         .where(db.ProcessingJob.recipe == "em-spa-preprocess")

src/murfey/server/__init__.py

@@ -181,6 +181,7 @@ def get_job_ids(tilt_series_id: int, appid: int) -> JobIDs:
         .where(db.AutoProcProgram.id == appid)
         .where(db.ProcessingJob.dc_id == db.DataCollection.id)
         .where(db.DataCollectionGroup.id == db.DataCollection.dcg_id)
+        .where(db.DataCollectionGroup.tag == db.DataCollection.dcg_tag)
         .where(db.Session.id == db.TiltSeries.session_id)
     ).all()
     return JobIDs(
@@ -2399,6 +2400,7 @@ def feedback_callback(header: dict, message: dict) -> None:
                     db.AutoProcProgram,
                 )
                 .where(db.DataCollection.dcg_id == db.DataCollectionGroup.id)
+                .where(db.DataCollection.dcg_tag == db.DataCollectionGroup.tag)
                 .where(db.ProcessingJob.dc_id == db.DataCollection.id)
                 .where(db.AutoProcProgram.pj_id == db.ProcessingJob.id)
                 .where(db.AutoProcProgram.id == message["program_id"])
@@ -2598,6 +2600,7 @@ def feedback_callback(header: dict, message: dict) -> None:
             ).all()
             if dcg:
                 dcgid = dcg[0].id
+                dcg_tag = dcg[0].tag
                 # flush_data_collections(message["source"], murfey_db)
             else:
                 logger.warning(
@@ -2610,13 +2613,15 @@ def feedback_callback(header: dict, message: dict) -> None:
                 select(db.DataCollection)
                 .where(db.DataCollection.tag == message.get("tag"))
                 .where(db.DataCollection.dcg_id == dcgid)
+                .where(db.DataCollection.dcg_tag == dcg_tag)
             ).all():
                 dcid = dc_murfey[0].id
             else:
                 if ispyb_session_id is None:
                     murfey_dc = db.DataCollection(
                         tag=message.get("tag"),
                         dcg_id=dcgid,
+                        dcg_tag=dcg_tag,
                     )
                 else:
                     record = DataCollection(
@@ -2649,6 +2654,7 @@ def feedback_callback(header: dict, message: dict) -> None:
                         id=dcid,
                         tag=message.get("tag"),
                         dcg_id=dcgid,
+                        dcg_tag=dcg_tag,
                     )
                 murfey_db.add(murfey_dc)
                 murfey_db.commit()
@@ -2666,6 +2672,7 @@ def feedback_callback(header: dict, message: dict) -> None:
             dc = murfey_db.exec(
                 select(db.DataCollection, db.DataCollectionGroup)
                 .where(db.DataCollection.dcg_id == db.DataCollectionGroup.id)
+                .where(db.DataCollection.dcg_tag == db.DataCollectionGroup.tag)
                 .where(db.DataCollectionGroup.session_id == murfey_session_id)
                 .where(db.DataCollectionGroup.tag == message["source"])
                 .where(db.DataCollection.tag == message["tag"])

src/murfey/server/api/__init__.py

@@ -556,6 +556,7 @@ def register_tomo_proc_params(
         .where(DataCollectionGroup.tag == proc_params.tag)
         .where(DataCollection.tag == proc_params.tilt_series_tag)
         .where(DataCollection.dcg_id == DataCollectionGroup.id)
+        .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
         .where(ProcessingJob.recipe == "em-tomo-preprocess")
@@ -685,6 +686,7 @@ def register_completed_tilt_series(
                 .where(DataCollectionGroup.tag == tilt_series_group.source)
                 .where(DataCollection.tag == ts.tag)
                 .where(DataCollection.dcg_id == DataCollectionGroup.id)
+                .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
                 .where(ProcessingJob.dc_id == DataCollection.id)
                 .where(AutoProcProgram.pj_id == ProcessingJob.id)
                 .where(ProcessingJob.recipe == "em-tomo-align")
@@ -987,6 +989,7 @@ async def request_spa_preprocessing(
             .where(DataCollectionGroup.session_id == session_id)
             .where(DataCollectionGroup.tag == proc_file.tag)
             .where(DataCollection.dcg_id == DataCollectionGroup.id)
+            .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
             .where(ProcessingJob.dc_id == DataCollection.id)
             .where(AutoProcProgram.pj_id == ProcessingJob.id)
             .where(ProcessingJob.recipe == "em-spa-preprocess")
@@ -1136,6 +1139,7 @@ async def request_tomography_preprocessing(
         .where(DataCollectionGroup.tag == proc_file.group_tag)
         .where(DataCollection.tag == proc_file.tag)
         .where(DataCollection.dcg_id == DataCollectionGroup.id)
+        .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
         .where(ProcessingJob.recipe == recipe_name)
@@ -1738,6 +1742,7 @@ def register_processing_success_in_ispyb(
         select(DataCollectionGroup, DataCollection, ProcessingJob, AutoProcProgram)
         .where(DataCollectionGroup.session_id == session_id)
         .where(DataCollection.dcg_id == DataCollectionGroup.id)
+        .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
     ).all()
@@ -1777,6 +1782,7 @@ def remove_session(client_id: int, db=murfey_db):
         select(DataCollectionGroup, DataCollection, ProcessingJob)
         .where(DataCollectionGroup.session_id == session_id)
         .where(DataCollection.dcg_id == DataCollectionGroup.id)
+        .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
     ).all()
     for c in collected_ids:

src/murfey/server/demo_api.py

@@ -386,6 +386,7 @@ def get_spa_proc_param_details(
         )
         .where(DataCollectionGroup.session_id == session_id)
         .where(DataCollectionGroup.id == DataCollection.dcg_id)
+        .where(DataCollectionGroup.tag == DataCollection.dcg_tag)
         .where(DataCollection.id == ProcessingJob.dc_id)
         .where(SPARelionParameters.pj_id == ProcessingJob.id)
         .where(SPAFeedbackParameters.pj_id == ProcessingJob.id)
@@ -437,6 +438,7 @@ def register_spa_proc_params(
             .where(DataCollectionGroup.session_id == session_id)
             .where(DataCollectionGroup.tag == proc_params.tag)
             .where(DataCollection.dcg_id == DataCollectionGroup.id)
+            .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
             .where(ProcessingJob.dc_id == DataCollection.id)
             .where(AutoProcProgram.pj_id == ProcessingJob.id)
             .where(ProcessingJob.recipe == "em-spa-preprocess")
@@ -497,6 +499,7 @@ def register_tomo_preproc_params(
         .where(DataCollectionGroup.tag == proc_params.tag)
         .where(DataCollection.tag == proc_params.tilt_series_tag)
         .where(DataCollection.dcg_id == DataCollectionGroup.id)
+        .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
         .where(ProcessingJob.recipe == "em-tomo-preprocess")
@@ -551,6 +554,7 @@ def register_tomo_proc_params(
         .where(DataCollectionGroup.tag == proc_params.tag)
         .where(DataCollection.tag == proc_params.tilt_series_tag)
         .where(DataCollection.dcg_id == DataCollectionGroup.id)
+        .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
         .where(ProcessingJob.recipe == "em-tomo-preprocess")
@@ -943,6 +947,7 @@ def flush_spa_processing(
         .where(DataCollectionGroup.session_id == session_id)
         .where(DataCollectionGroup.tag == tag.tag)
         .where(DataCollection.dcg_id == DataCollectionGroup.id)
+        .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
         .where(ProcessingJob.recipe == "em-spa-preprocess")
@@ -1073,6 +1078,7 @@ async def request_spa_preprocessing(
             .where(DataCollectionGroup.session_id == session_id)
             .where(DataCollectionGroup.tag == proc_file.tag)
             .where(DataCollection.dcg_id == DataCollectionGroup.id)
+            .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
             .where(ProcessingJob.dc_id == DataCollection.id)
             .where(AutoProcProgram.pj_id == ProcessingJob.id)
             .where(ProcessingJob.recipe == "em-spa-preprocess")
@@ -1109,6 +1115,7 @@ async def request_spa_preprocessing(
             )
             .where(DataCollectionGroup.tag == proc_file.tag)
             .where(DataCollection.dcg_id == DataCollectionGroup.id)
+            .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
             .where(ProcessingJob.dc_id == DataCollection.id)
             .where(AutoProcProgram.pj_id == ProcessingJob.id)
             .where(ProcessingJob.recipe == "em-spa-preprocess")
@@ -1230,6 +1237,7 @@ async def request_tomography_preprocessing(
         select(DataCollectionGroup, DataCollection, ProcessingJob, AutoProcProgram)
         .where(DataCollectionGroup.session_id == session_id)
         .where(DataCollectionGroup.id == DataCollection.dcg_id)
+        .where(DataCollectionGroup.tag == DataCollection.dcg_tag)
         .where(DataCollection.tag == proc_file.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
@@ -1656,6 +1664,7 @@ def remove_session(client_id: int, db=murfey_db):
         select(DataCollectionGroup, DataCollection, ProcessingJob)
         .where(DataCollectionGroup.session_id == session_id)
         .where(DataCollection.dcg_id == DataCollectionGroup.id)
+        .where(DataCollection.dcg_tag == DataCollectionGroup.tag)
         .where(ProcessingJob.dc_id == DataCollection.id)
     ).all()
     for c in collected_ids:

src/murfey/util/db.py

@@ -6,7 +6,7 @@
 from typing import List, Optional
 
 import sqlalchemy
-from sqlmodel import Field, Relationship, SQLModel, create_engine
+from sqlmodel import Field, Relationship, SQLModel, UniqueConstraint, create_engine
 
 """
 GENERAL
@@ -348,6 +348,8 @@ class Tilt(SQLModel, table=True):  # type: ignore
 
 
 class DataCollectionGroup(SQLModel, table=True):  # type: ignore
+    __table_args__ = (UniqueConstraint("id", "tag", name="dcg_constraint"),)
+
     id: int = Field(primary_key=True)
     session_id: int = Field(foreign_key="session.id", primary_key=True)
     tag: str = Field(primary_key=True)
@@ -372,6 +374,7 @@ class DataCollection(SQLModel, table=True):  # type: ignore
     id: int = Field(primary_key=True, unique=True)
     tag: str = Field(primary_key=True)
     dcg_id: int = Field(foreign_key="datacollectiongroup.id")
+    dcg_tag: str = Field(foreign_key="datacollectiongroup.tag")
     data_collection_group: Optional[DataCollectionGroup] = Relationship(
         back_populates="data_collections"
     )