Merge branch 'main' into chore/bump-uv

Author: Hoeze

packages/grz-cli/pyproject.toml

@@ -23,7 +23,6 @@ dependencies = [
     "pydantic-settings >=2.11,<3",
     "platformdirs >=4.3.6,<5",
     "grz-pydantic-models >=2.5,<3",
-    "pysam ==0.23.*",
     "rich ==13.*",
     "requests >=2.32.3,<3",
     "grz-common >=1.5.0,<2",

packages/grz-common/pyproject.toml

@@ -25,7 +25,6 @@ dependencies = [
     "platformdirs >=4.3.6,<5",
     "grz-pydantic-models >=2.2.0",
     "grz-check >=0.2.1",
-    "pysam ==0.23.*",
     "rich ==13.*",
     "requests >=2.32.3,<3",
 ]

packages/grz-common/src/grz_common/progress/states.py

@@ -11,6 +11,7 @@ class State(TypedDict, total=False):
     """
 
     errors: list[str]
+    submission_id: str
 
 
 class ValidationState(State):

packages/grz-common/src/grz_common/workers/download.py

@@ -164,6 +164,7 @@ def download_file(
         s3_object_id: str,
         progress_logger: FileProgressLogger[DownloadState],
         file_metadata: SubmissionFileMetadata,
+        submission_id: str,
     ):
         """
         Download a single file from S3 to the specified local_file_path.
@@ -179,7 +180,11 @@ def download_file(
             self._download_with_progress(str(local_file_path), s3_object_id)
 
             self.__log.info(f"Download complete for {str(local_file_path)}.")
-            progress_logger.set_state(local_file_path, file_metadata, state=DownloadState(download_successful=True))
+            progress_logger.set_state(
+                local_file_path,
+                file_metadata,
+                state=DownloadState(download_successful=True, submission_id=submission_id),
+            )
 
         except botocore.exceptions.ClientError as e:
             if e.response.get("Error", {}).get("Code") == "404":
@@ -190,13 +195,17 @@ def download_file(
                 exc = e  # type: ignore[assignment]
             self.__log.error(error_msg)
             progress_logger.set_state(
-                local_file_path, file_metadata, state=DownloadState(download_successful=False, errors=[str(exc)])
+                local_file_path,
+                file_metadata,
+                state=DownloadState(download_successful=False, errors=[str(exc)], submission_id=submission_id),
             )
             raise exc from e
         except Exception as e:
             self.__log.error("Download failed for '%s': %s", str(local_file_path), e)
             progress_logger.set_state(
-                local_file_path, file_metadata, state=DownloadState(download_successful=False, errors=[str(e)])
+                local_file_path,
+                file_metadata,
+                state=DownloadState(download_successful=False, errors=[str(e)], submission_id=submission_id),
             )
             raise e
 
@@ -217,7 +226,11 @@ def download(self, submission_id: str, encrypted_submission: EncryptedSubmission
             file_key = f"{submission_id}/files/{relative_encrypted_path}"
 
             logged_state = progress_logger.get_state(local_file_path, file_metadata)
-            if logged_state and logged_state.get("download_successful"):
+            if (
+                logged_state
+                and logged_state.get("download_successful")
+                and logged_state.get("submission_id") == submission_id
+            ):
                 self.__log.info(
                     "File '%s' already downloaded (at '%s'), skipping.",
                     file_key,
@@ -226,7 +239,7 @@ def download(self, submission_id: str, encrypted_submission: EncryptedSubmission
                 continue
 
             self.__log.info("Downloading file: '%s' -> '%s'", file_key, str(local_file_path))
-            self.download_file(local_file_path, file_key, progress_logger, file_metadata)
+            self.download_file(local_file_path, file_key, progress_logger, file_metadata, submission_id)
 
 
 class InboxSubmissionState(enum.StrEnum):

packages/grz-common/src/grz_common/workers/submission.py

@@ -259,6 +259,10 @@ def __init__(self, metadata_dir: str | PathLike, files_dir: str | PathLike):
 
         self.metadata = SubmissionMetadata(self.metadata_dir / "metadata.json")
 
+    @property
+    def submission_id(self) -> str:
+        return self.metadata.content.submission_id
+
     @property
     def files(self) -> dict[Path, SubmissionFileMetadata]:
         """
@@ -312,14 +316,19 @@ def validate_files(  # noqa: C901, PLR0912, PLR0915
         tasks = []
 
         def should_check_file(file_path: Path, file_metadata: SubmissionFileMetadata) -> bool:
-            # Check against both logs. If either is missing a "pass", re-check.
             checksum_state = checksum_progress_logger.get_state(file_path, file_metadata)
             seq_data_state = seq_data_progress_logger.get_state(file_path, file_metadata)
             is_seq_file = file_metadata.file_type in ("fastq", "bam")
-
-            checksum_passed = checksum_state and checksum_state.get("validation_passed")
-            seq_data_passed = seq_data_state and seq_data_state.get("validation_passed")
-
+            checksum_passed = (
+                checksum_state
+                and checksum_state.get("validation_passed")
+                and checksum_state.get("submission_id") == self.submission_id
+            )
+            seq_data_passed = (
+                seq_data_state
+                and seq_data_state.get("validation_passed")
+                and seq_data_state.get("submission_id") == self.submission_id
+            )
             if is_seq_file:
                 return not (checksum_passed and seq_data_passed)
             return not checksum_passed
@@ -448,11 +457,15 @@ def _execute_task(task_type, paths, metas, kwargs, pbar):
                         checksum_issues.append("File not found for size check.")
 
                     checksum_passed = not checksum_issues
-                    checksum_state = ValidationState(errors=checksum_issues, validation_passed=checksum_passed)
+                    checksum_state = ValidationState(
+                        errors=checksum_issues, validation_passed=checksum_passed, submission_id=self.submission_id
+                    )
                     checksum_progress_logger.set_state(file_path, file_metadata, checksum_state)
 
                     if file_metadata.file_type in ("fastq", "bam"):
-                        seq_data_state = ValidationState(errors=report.errors, validation_passed=report.is_valid)
+                        seq_data_state = ValidationState(
+                            errors=report.errors, validation_passed=report.is_valid, submission_id=self.submission_id
+                        )
                         seq_data_progress_logger.set_state(file_path, file_metadata, seq_data_state)
 
                 if not no_mmap:
@@ -524,6 +537,7 @@ def encrypt(
                 (logged_state is None)
                 or not logged_state.get("encryption_successful", False)
                 or not encrypted_file_path.is_file()
+                or logged_state.get("submission_id") != self.submission_id
             ):
                 self.__log.info(
                     "Encrypting file: '%s' -> '%s'",
@@ -543,15 +557,17 @@ def encrypt(
                     progress_logger.set_state(
                         file_path,
                         file_metadata,
-                        state=EncryptionState(encryption_successful=True),
+                        state=EncryptionState(encryption_successful=True, submission_id=self.submission_id),
                     )
                 except Exception as e:
                     self.__log.error("Encryption failed for '%s'", str(file_path))
 
                     progress_logger.set_state(
                         file_path,
                         file_metadata,
-                        state=EncryptionState(encryption_successful=False, errors=[str(e)]),
+                        state=EncryptionState(
+                            encryption_successful=False, errors=[str(e)], submission_id=self.submission_id
+                        ),
                     )
 
                     raise e
@@ -704,6 +720,7 @@ def decrypt(
             if (
                 (logged_state is None)
                 or not logged_state.get("decryption_successful", False)
+                or logged_state.get("submission_id") != self.submission_id
                 or not decrypted_file_path.is_file()
             ):
                 self.__log.info(
@@ -719,15 +736,17 @@ def decrypt(
                     progress_logger.set_state(
                         encrypted_file_path,
                         file_metadata,
-                        state=DecryptionState(decryption_successful=True),
+                        state=DecryptionState(decryption_successful=True, submission_id=self.submission_id),
                     )
                 except Exception as e:
                     self.__log.error("Decryption failed for '%s'", str(encrypted_file_path))
 
                     progress_logger.set_state(
                         encrypted_file_path,
                         file_metadata,
-                        state=DecryptionState(decryption_successful=False, errors=[str(e)]),
+                        state=DecryptionState(
+                            decryption_successful=False, errors=[str(e)], submission_id=self.submission_id
+                        ),
                     )
 
                     raise e

packages/grz-common/src/grz_common/workers/upload.py

@@ -168,8 +168,11 @@ def _upload_logged_files(self, encrypted_submission, progress_logger, files_to_u
             self.__log.debug("state for %s: %s", file_path, logged_state)
 
             s3_object_id = files_to_upload[file_path]
-
-            if (logged_state is None) or not logged_state.get("upload_successful", False):
+            if (
+                (logged_state is None)
+                or not logged_state.get("upload_successful", False)
+                or logged_state.get("submission_id") != encrypted_submission.submission_id
+            ):
                 self.__log.info(
                     "Uploading file: '%s' -> '%s'",
                     str(file_path),
@@ -183,22 +186,25 @@ def _upload_logged_files(self, encrypted_submission, progress_logger, files_to_u
                     progress_logger.set_state(
                         file_path,
                         file_metadata,
-                        state=UploadState(upload_successful=True),
+                        state=UploadState(upload_successful=True, submission_id=encrypted_submission.submission_id),
                     )
                 except Exception as e:
                     self.__log.error("Upload failed for '%s'", str(file_path))
 
                     progress_logger.set_state(
                         file_path,
                         file_metadata,
-                        state=UploadState(upload_successful=False, errors=[str(e)]),
+                        state=UploadState(
+                            upload_successful=False, errors=[str(e)], submission_id=encrypted_submission.submission_id
+                        ),
                     )
 
                     raise e
             else:
                 self.__log.info(
-                    "File '%s' already uploaded (at '%s')",
+                    "File '%s' already uploaded for submission '%s' (at '%s')",
                     str(file_path),
+                    encrypted_submission.submission_id,
                     str(s3_object_id),
                 )
 

packages/grz-common/src/grz_common/workers/worker.py

@@ -370,13 +370,13 @@ def populate(self, s3_options: S3Options, db: SubmissionDb, submission_id: str,
         submission_diff, donors_diff = db.diff(submission_id, metadata, submission_date)
 
         # check for destructive changes
-        if not force and (len(submission_diff.deleted) + len(submission_diff.updated)) > 0:
+        if not force and submission_diff.has_pending_destructive:
             raise RuntimeError(
                 f"Would update/delete existing submission data in the database, but `force` not set. "
                 f"submission_id={submission_id!r}, submission_diff={submission_diff!r}"
             )
 
-        if not force and (len(donors_diff.deleted) + len(donors_diff.updated)) > 0:
+        if not force and donors_diff.has_pending_destructive:
             raise RuntimeError(
                 f"Would update/delete existing donors in the database, but `force` not set. "
                 f"submission_id={submission_id!r}, donors_diff={donors_diff!r}"

packages/grz-db/src/grz_db/migrations/versions/c5f6a8b9d2e1_add_grzctl_and_qc_workflow_versions.py

@@ -0,0 +1,35 @@
+"""add grzctl_version and qc_workflow_version columns
+
+Revision ID: c5f6a8b9d2e1
+Revises: 834bd50b8734
+Create Date: 2026-05-06 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "c5f6a8b9d2e1"
+down_revision: str | Sequence[str] | None = "834bd50b8734"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # Add grzctl_versions to submission_states
+    op.add_column(
+        "submission_states",
+        sa.Column("grzctl_versions", sa.JSON(), nullable=True),
+    )
+
+    # Add qc_workflow_version to detailed_qc_results
+    op.add_column(
+        "detailed_qc_results",
+        sa.Column("qc_workflow_version", sa.String(length=64), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    raise RuntimeError("Downgrades not supported.")

packages/grz-db/src/grz_db/models/submission/__init__.py

@@ -257,6 +257,11 @@ class SubmissionStateLogBase(SQLModel):
 
     state: SubmissionStateEnum
     data: dict[str, Any] | None = Field(default=None, sa_column=Column(JSON))
+    grzctl_versions: dict[str, str] | None = Field(
+        default=None,
+        description="grzctl versions that created this state log (nullable for backward compatibility with old state logs)",
+        sa_column=Column(JSON),
+    )
     timestamp: datetime.datetime = Field(
         default_factory=lambda: datetime.datetime.now(datetime.UTC),
         sa_column=Column(DateTime(timezone=True), nullable=False),
@@ -465,6 +470,11 @@ class DetailedQCResult(SQLModel, table=True):
     targeted_regions_above_min_coverage: float
     targeted_regions_above_min_coverage_passed_qc: bool
     targeted_regions_above_min_coverage_percent_deviation: float
+    qc_workflow_version: str | None = Field(
+        default=None,
+        sa_column=Column(sa.String(length=64), nullable=True),
+        description="QC workflow version (nullable for backward compatibility with old results)",
+    )
 
     model_config = ConfigDict(  # type: ignore
         populate_by_name=True,
@@ -737,6 +747,7 @@ def update_submission_state(
         submission_id: str,
         state: SubmissionStateEnum,
         data: dict | None = None,
+        grzctl_versions: dict[str, str] | None = None,
     ) -> SubmissionStateLog:
         """
         Updates a submission's state to the specified state.
@@ -745,6 +756,7 @@ def update_submission_state(
             submission_id: Submission ID of the submission to update.
             state: New state of the submission.
             data: Optional data to attach to the update.
+            grzctl_versions: Optional dictionary of grzctl dependency versions.
 
         Returns:
             An instance of SubmissionStateLog.
@@ -757,7 +769,11 @@ def update_submission_state(
                 raise ValueError("No author defined")
 
             state_log_payload = SubmissionStateLogPayload(
-                submission_id=submission_id, author_name=self._author.name, state=state, data=data
+                submission_id=submission_id,
+                author_name=self._author.name,
+                state=state,
+                data=data,
+                grzctl_versions=grzctl_versions,
             )
             signature = state_log_payload.sign(self._author.private_key())
 
@@ -919,6 +935,24 @@ def get_submission(self, submission_id: str) -> Submission | None:
             submission = session.exec(statement).first()
             return submission
 
+    def get_submissions(self, submission_ids: Sequence[str]) -> list[Submission | None]:
+        """Fetch a specific set of submissions by their IDs in a single query.
+
+        :param submission_ids: IDs to fetch.
+        :returns: A list of the same length as *submission_ids*, where each element is the
+                  matching :class:`Submission` or ``None`` when the ID does not exist.
+        """
+        if not submission_ids:
+            return []
+        with self._get_session() as session:
+            statement = (
+                select(Submission)
+                .where(Submission.id.in_(submission_ids))  # type: ignore[attr-defined]
+                .options(selectinload(Submission.states))  # type: ignore[arg-type]
+            )
+            found = {s.id: s for s in session.exec(statement).all()}
+        return [found.get(sid) for sid in submission_ids]
+
     def list_submissions(
         self,
         limit: int | None,