refactor(sp41): align submission_dedup with cyclone.store conventions (main Base, db.SessionLocal, now(timezone.utc), exceptions.py)

This commit is contained in:
Nora
2026-07-07 22:16:07 -06:00
parent 0e130579b4
commit 53df3f0684
4 changed files with 87 additions and 55 deletions
+16
View File
@@ -982,3 +982,19 @@ class Resubmission(Base):
unique=True, unique=True,
), ),
) )
class SubmissionRecord(Base):
"""SP41: one row per (claim_id, submitted_at) pre-flight dedup guard.
The 30-day dedup window is enforced by ``cyclone.store.submission_dedup``.
This table records (claim_id, submitted_at) for every push that passed
the pre-flight check. Past the window, the prior record is ignored
(the guard lets through re-submits older than
``cyclone.store.submission_dedup.DEFAULT_WINDOW_DAYS``).
"""
__tablename__ = "submission_dedup"
claim_id: Mapped[str] = mapped_column(String(64), primary_key=True)
submitted_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, index=True)
+25
View File
@@ -4,6 +4,8 @@ These are part of the public API — callers (API endpoints) catch them
and translate to HTTP 409 Conflict responses. and translate to HTTP 409 Conflict responses.
""" """
from datetime import datetime
class AlreadyMatchedError(Exception): class AlreadyMatchedError(Exception):
"""Raised by ``CycloneStore.manual_match`` when the claim is already paired. """Raised by ``CycloneStore.manual_match`` when the claim is already paired.
@@ -38,4 +40,27 @@ class InvalidStateError(Exception):
self.activity_kind = activity_kind self.activity_kind = activity_kind
super().__init__( super().__init__(
f"invalid state {current_state} for apply (kind={activity_kind})" f"invalid state {current_state} for apply (kind={activity_kind})"
)
class DuplicateClaimError(Exception):
"""Raised by ``CycloneStore.check_submission_duplicate`` when a claim_id
is re-submitted within the 30-day dedup window (SP41).
Mirrors the other 409-class exceptions: callers catch it at the API
boundary and surface it as a 409 Conflict. Carries
``original_submission_at`` so the UI can render "already submitted on
YYYY-MM-DD" without re-querying.
Note: the "30-day" literal in the message mirrors
``cyclone.store.submission_dedup.DEFAULT_WINDOW_DAYS`` — keep them in
sync if either ever moves.
"""
def __init__(self, claim_id: str, original_submission_at: datetime):
self.claim_id = claim_id
self.original_submission_at = original_submission_at
super().__init__(
f"claim_id {claim_id!r} was already submitted at "
f"{original_submission_at.isoformat()}; within 30-day window"
) )
+44 -53
View File
@@ -1,74 +1,65 @@
"""Claim-id dedup at the SFTP pre-flight stage. """Claim-id dedup at the SFTP pre-flight stage.
Schema: a ``submission_dedup`` table records (claim_id, submitted_at). A Schema: the ``submission_dedup`` table (defined on the main ``Base`` in
re-submission of the same claim_id within ``DEFAULT_WINDOW_DAYS`` (30) ``cyclone.db``) records (claim_id, submitted_at). A re-submission of
is rejected with ``DuplicateClaimError``. Past the window, the guard the same claim_id within ``DEFAULT_WINDOW_DAYS`` (30) is rejected with
lets it through — the operator may be retrying a 999-rejected file ``DuplicateClaimError``. Past the window, the guard lets it through —
from >30 days ago, which is a legitimate rebuild path. the operator may be retrying a 999-rejected file from >30 days ago,
which is a legitimate rebuild path.
""" """
from __future__ import annotations from __future__ import annotations
from datetime import datetime, timedelta from datetime import datetime, timedelta, timezone
from sqlalchemy import Column, DateTime, String, create_engine, insert, select from sqlalchemy import select
from sqlalchemy.orm import declarative_base
from cyclone import db as cycl_db
from cyclone.db import SubmissionRecord
from cyclone.store.exceptions import DuplicateClaimError
DEFAULT_WINDOW_DAYS = 30 DEFAULT_WINDOW_DAYS = 30
_Base = declarative_base()
class _SubmissionRecord(_Base):
__tablename__ = "submission_dedup"
claim_id = Column(String(64), primary_key=True)
submitted_at = Column(DateTime, nullable=False, index=True)
class DuplicateClaimError(Exception):
def __init__(self, claim_id: str, original_submission_at: datetime):
super().__init__(
f"claim_id {claim_id!r} was already submitted at "
f"{original_submission_at.isoformat()}; within {DEFAULT_WINDOW_DAYS}-day window"
)
self.claim_id = claim_id
self.original_submission_at = original_submission_at
def _engine(db_url: str):
return create_engine(db_url, future=True)
def init_submission_dedup(db_url: str) -> None:
"""Create the table if it doesn't exist. Idempotent."""
_Base.metadata.create_all(_engine(db_url))
def record_submission(claim_id: str, submitted_at: datetime, db_url: str) -> None: def record_submission(claim_id: str, submitted_at: datetime, db_url: str) -> None:
init_submission_dedup(db_url) """Upsert (claim_id, submitted_at). Idempotent on claim_id.
eng = _engine(db_url)
with eng.begin() as conn: ``db_url`` is accepted for back-compat with the original Task 8
conn.execute(insert(_SubmissionRecord).values( signature but ignored — sessions now flow through the process-wide
claim_id=claim_id, submitted_at=submitted_at, ``cycl_db.SessionLocal()`` factory that ``db.init_db()`` set up.
).prefix_with("OR REPLACE")) """
del db_url # signature-preserving; conftest wires the per-test DB
with cycl_db.SessionLocal()() as session:
session.merge(SubmissionRecord(
claim_id=claim_id,
submitted_at=submitted_at,
))
session.commit()
def check_duplicate( def check_duplicate(
claim_id: str, claim_id: str,
db_url: str, db_url: str | None = None,
now: datetime | None = None, now: datetime | None = None,
window_days: int = DEFAULT_WINDOW_DAYS, window_days: int = DEFAULT_WINDOW_DAYS,
) -> None: ) -> None:
"""Raise DuplicateClaimError if claim_id was submitted within the window.""" """Raise ``DuplicateClaimError`` if ``claim_id`` was submitted within the window.
init_submission_dedup(db_url)
``db_url`` is accepted for back-compat with the original Task 8
signature but ignored — sessions flow through ``cycl_db.SessionLocal()``
just like the rest of ``cyclone.store``. Tests that pass an explicit
``db_url`` keep working because ``cycl_db.SessionLocal()()`` points at
the URL the conftest resolved under ``CYCLONE_DB_URL``.
"""
del db_url # signature-preserving
if now is None: if now is None:
now = datetime.utcnow() now = datetime.now(timezone.utc)
cutoff = now - timedelta(days=window_days) cutoff = now - timedelta(days=window_days)
eng = _engine(db_url) with cycl_db.SessionLocal()() as session:
with eng.connect() as conn: row = session.scalars(
stmt = select(_SubmissionRecord).where( select(SubmissionRecord).where(
_SubmissionRecord.claim_id == claim_id, SubmissionRecord.claim_id == claim_id,
_SubmissionRecord.submitted_at >= cutoff, SubmissionRecord.submitted_at >= cutoff,
) )
row = conn.execute(stmt).first() ).first()
if row is not None: if row is not None:
raise DuplicateClaimError(claim_id, row.submitted_at) raise DuplicateClaimError(claim_id, row.submitted_at)
+2 -2
View File
@@ -1,7 +1,7 @@
"""Claim-id dedup at the SFTP pre-flight stage.""" """Claim-id dedup at the SFTP pre-flight stage."""
from datetime import datetime from datetime import datetime
from cyclone.store.exceptions import DuplicateClaimError
from cyclone.store.submission_dedup import ( from cyclone.store.submission_dedup import (
DuplicateClaimError,
record_submission, record_submission,
check_duplicate, check_duplicate,
DEFAULT_WINDOW_DAYS, DEFAULT_WINDOW_DAYS,
@@ -35,4 +35,4 @@ def test_check_duplicate_passes_after_window(tmp_path):
def test_default_window_is_30_days(): def test_default_window_is_30_days():
assert DEFAULT_WINDOW_DAYS == 30 assert DEFAULT_WINDOW_DAYS == 30