fix(grz-tools): map duplicate tanG and upload errors correctly

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

@@ -16,6 +16,7 @@
 
 import botocore.handlers
 from boto3.s3.transfer import S3Transfer, TransferConfig  # type: ignore[import-untyped]
+from grz_common.exceptions import UploadError
 from grz_pydantic_models.submission.metadata import REDACTED_TAN
 from tqdm.auto import tqdm
 
@@ -37,12 +38,6 @@
 botocore.handlers.VALID_BUCKET = re.compile(r"^[:a-zA-Z0-9.\-_]{1,255}$")  # type: ignore[import-untyped]
 
 
-class UploadError(Exception):
-    """Exception raised when an upload fails"""
-
-    pass
-
-
 class UploadWorker(metaclass=abc.ABCMeta):
     """Worker baseclass for uploading encrypted submissions"""
 

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

@@ -10,14 +10,15 @@
     DecryptionError,
     EncryptionError,
     IncompleteSubmissionError,
+    UploadError,
 )
 
 from ..models.identifiers import IdentifiersModel
 from ..models.s3 import S3Options
 from ..progress import EncryptionState, FileProgressLogger, ValidationState
 from .download import S3BotoDownloadWorker
 from .submission import EncryptedSubmission, Submission, SubmissionValidationError
-from .upload import S3BotoUploadWorker, UploadError
+from .upload import S3BotoUploadWorker
 
 log = logging.getLogger(__name__)
 

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

@@ -634,6 +634,19 @@ def add_submission(
                 session.rollback()
                 raise
 
+    @staticmethod
+    def _is_tan_g_unique_violation(e: IntegrityError) -> bool:
+        # SQLite
+        if "UNIQUE constraint failed: submissions.tan_g" in str(e):
+            return True
+        # PostgreSQL (psycopg / psycopg2)
+        orig = getattr(e, "orig", None)
+        if orig is not None:
+            diag = getattr(orig, "diag", None)
+            if diag is not None:
+                return getattr(diag, "constraint_name", None) == "ix_submissions_tan_g"
+        return False
+
     def modify_submission(self, submission_id: str, key: str, value: Any) -> Submission:  # noqa: C901
         if key not in SubmissionBase.model_fields:
             raise ValueError(f"Unknown column key '{key}'")
@@ -667,7 +680,7 @@ def modify_submission(self, submission_id: str, key: str, value: Any) -> Submiss
                 return submission
             except IntegrityError as e:
                 session.rollback()
-                if "UNIQUE constraint failed: submissions.tanG" in str(e) and key == "tan_g":
+                if self._is_tan_g_unique_violation(e) and key == "tan_g":
                     raise DuplicateTanGError() from e
                 raise
             except Exception:
@@ -698,7 +711,7 @@ def update_submission(self, submission: Submission) -> Submission:
                 return db_submission
             except IntegrityError as e:
                 session.rollback()
-                if "UNIQUE constraint failed: submissions.tanG" in str(e):
+                if self._is_tan_g_unique_violation(e):
                     raise DuplicateTanGError() from e
                 raise
             except Exception:

packages/grz-db/tests/test_submission.py

@@ -3,6 +3,7 @@
 import pytest
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ed25519
+from grz_db.errors import DuplicateTanGError
 from grz_db.models.author import Author
 from grz_db.models.submission import Submission, SubmissionDb
 from grz_pydantic_models.submission.metadata import GrzSubmissionMetadata
@@ -11,6 +12,8 @@
 SUBMISSION_ID = "123456789_2024-01-01_abcdef01"
 SUBMISSION_ID_2 = "123456789_2024-01-02_abcdef02"
 SUBMISSION_ID_3 = "123456789_2024-01-03_abcdef03"
+TAN_G_1 = "a" * 64
+TAN_G_2 = "b" * 64
 
 
 @pytest.fixture(scope="function")
@@ -147,3 +150,25 @@ def test_get_submissions_includes_states(db: SubmissionDb) -> None:
 
     assert result[0] is not None
     assert len(result[0].states) >= 1
+
+
+def test_modify_submission_raises_on_duplicate_tan_g(db: SubmissionDb) -> None:
+    """Raise DuplicateTanGError when tan_g collides with an existing row."""
+    db.add_submission(SUBMISSION_ID)
+    db.modify_submission(SUBMISSION_ID, "tan_g", TAN_G_1)
+
+    db.add_submission(SUBMISSION_ID_2)
+    with pytest.raises(DuplicateTanGError):
+        db.modify_submission(SUBMISSION_ID_2, "tan_g", TAN_G_1)
+
+
+def test_update_submission_raises_on_duplicate_tan_g(db: SubmissionDb) -> None:
+    """Raise DuplicateTanGError when tan_g collides with an existing row."""
+    sub1 = db.add_submission(SUBMISSION_ID)
+    sub1.tan_g = TAN_G_1
+    db.update_submission(sub1)
+
+    sub2 = db.add_submission(SUBMISSION_ID_2)
+    sub2.tan_g = TAN_G_1  # same tan_g as sub1
+    with pytest.raises(DuplicateTanGError):
+        db.update_submission(sub2)