Ref #806: prevent duplicate entries in pending records table

ctms/crud.py

@@ -352,8 +352,9 @@ def schedule_acoustic_record(
     email_id: UUID4,
     metrics: Optional[Dict] = None,
 ) -> None:
-    db_pending_record = PendingAcousticRecord(email_id=email_id)
-    db.add(db_pending_record)
+    stmt = insert(PendingAcousticRecord).values(email_id=email_id)
+    stmt = stmt.on_conflict_do_nothing()
+    db.execute(stmt)
     if metrics:
         metrics["pending_acoustic_sync"].inc()
 

ctms/models.py

@@ -249,6 +249,8 @@ class PendingAcousticRecord(Base, TimestampMixin):
 
     email = relationship("Email", uselist=False)
 
+    __table_args__ = (UniqueConstraint("email_id", name="uix_pr_email_id"),)
+
 
 class StripeBase(Base):
     """Base class for Stripe objects."""

migrations/versions/20230915_375dd57cfed1_add_unicity_constraint_in_pending_.py

@@ -0,0 +1,32 @@
+"""Add unicity constraint in pending records
+
+Revision ID: 375dd57cfed1
+Revises: 8f93a0e590f0
+Create Date: 2023-09-15 10:11:28.618091
+
+"""
+# pylint: disable=no-member invalid-name
+# no-member is triggered by alembic.op, which has dynamically added functions
+# invalid-name is triggered by migration file names with a date prefix
+# invalid-name is triggered by top-level alembic constants like revision instead of REVISION
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "375dd57cfed1"  # pragma: allowlist secret
+down_revision = "8f93a0e590f0"  # pragma: allowlist secret
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_unique_constraint("uix_pr_email_id", "pending_acoustic", ["email_id"])
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint("uix_pr_email_id", "pending_acoustic", type_="unique")
+    # ### end Alembic commands ###

tests/unit/test_crud.py

@@ -124,6 +124,21 @@ def test_schedule_acoustic_record(dbsession, email_factory):
     assert dbsession.query(PendingAcousticRecord).one().email_id == email.email_id
 
 
+def test_schedule_acoustic_record_is_unique(dbsession, email_factory):
+    email = email_factory()
+    dbsession.commit()
+    schedule_acoustic_record(dbsession, email.email_id)
+    schedule_acoustic_record(dbsession, email.email_id)
+    dbsession.commit()
+
+    assert (
+        dbsession.query(PendingAcousticRecord)
+        .filter(PendingAcousticRecord.email_id == email.email_id)
+        .count()
+        == 1
+    )
+
+
 def test_get_acoustic_records_before_filter_by_end_time(
     dbsession, pending_acoustic_record_factory
 ):