Link FormSubmission to Input by foreign key (#639)

alembic/versions/003_formsubmission_input_id_fk.py

@@ -0,0 +1,42 @@
+"""formsubmission.input_id FK to inputs.
+
+Revision ID: 003
+Revises: 002
+Create Date: 2026-08-07
+
+Adds a nullable input_id FK on formsubmission -> inputs.input_id so submissions
+can link to the Input they were filled from, instead of only duplicating the
+transcript into input_text. input_text is kept for now (read by the
+/forms/submissions and analytics endpoints); dropping it is a deferred
+follow-up once those readers move to the FK.
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "003"
+down_revision: str | None = "002"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # batch_alter_table: SQLite can't ALTER a table to add a FK constraint in
+    # place (no ALTER-constraint support), so this goes through its
+    # copy-and-move strategy. On Postgres it emits a plain ALTER TABLE.
+    with op.batch_alter_table("formsubmission") as batch_op:
+        batch_op.add_column(sa.Column("input_id", sa.Uuid(), nullable=True))
+        batch_op.create_foreign_key(
+            "fk_formsubmission_input_id_inputs",
+            "inputs",
+            ["input_id"],
+            ["input_id"],
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("formsubmission") as batch_op:
+        batch_op.drop_constraint("fk_formsubmission_input_id_inputs", type_="foreignkey")
+        batch_op.drop_column("input_id")

app/api/routes/forms.py

@@ -59,7 +59,10 @@ def fill_form(form: FormFill, db: Session = Depends(get_db)):
         )
 
         submission = FormSubmission(
-            template_id=form.template_id, input_text=transcript, output_pdf_path=path
+            template_id=form.template_id,
+            input_id=form.input_id,
+            input_text=transcript,
+            output_pdf_path=path,
         )
         return create_form(db, submission)
     except Exception as e:

app/api/routes/jobs.py

@@ -44,7 +44,7 @@ def submit_async_form_fill(form: AsyncFormFill, db: Session = Depends(get_db)):
 
     jobs: list[AsyncJobSubmitResponse] = []
     for tid in form.template_ids:
-        result = fill_form_task.delay(tid, transcript, form.model)
+        result = fill_form_task.delay(tid, transcript, str(form.input_id), form.model)
         job = Job(
             celery_task_id=result.id,
             job_type="form_generation",

app/models/models.py

@@ -30,6 +30,7 @@ class Template(SQLModel, table=True):
 class FormSubmission(SQLModel, table=True):
     id: int | None = Field(default=None, primary_key=True)
     template_id: int = Field(foreign_key="template.id")
+    input_id: UUID | None = Field(default=None, foreign_key="inputs.input_id")
     input_text: str
     output_pdf_path: str
     created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

app/tasks/fill.py

@@ -1,5 +1,6 @@
 import logging
 from datetime import datetime, timezone
+from uuid import UUID
 
 from app.core.celery import celery_app
 from app.db.database import get_session
@@ -11,7 +12,7 @@
 
 
 @celery_app.task(bind=True, name="fill_form")
-def fill_form_task(self, template_id: int, input_text: str, model: str | None = None):
+def fill_form_task(self, template_id: int, input_text: str, input_id_str: str, model: str | None = None):
     session = next(get_session())
     try:
         job = get_job_by_celery_id(session, self.request.id)
@@ -37,6 +38,7 @@ def fill_form_task(self, template_id: int, input_text: str, model: str | None =
 
         submission = FormSubmission(
             template_id=template_id,
+            input_id=UUID(input_id_str),
             input_text=input_text,
             output_pdf_path=path,
         )

tests/test_api.py

@@ -44,8 +44,16 @@ def test_form_submission_roundtrip(self, db):
         db.commit()
         db.refresh(tpl)
 
+        input_record = Input(
+            input_type=InputType.text, status=InputStatus.ready, transcript="John Doe, firefighter"
+        )
+        db.add(input_record)
+        db.commit()
+        db.refresh(input_record)
+
         sub = FormSubmission(
             template_id=tpl.id,
+            input_id=input_record.input_id,
             input_text="John Doe, firefighter",
             output_pdf_path="src/outputs/filled.pdf",
         )
@@ -56,6 +64,7 @@ def test_form_submission_roundtrip(self, db):
         fetched = db.get(FormSubmission, sub.id)
         assert fetched is not None
         assert fetched.template_id == tpl.id
+        assert fetched.input_id == input_record.input_id
         assert fetched.input_text == "John Doe, firefighter"
         assert fetched.created_at is not None
 
@@ -208,6 +217,9 @@ def test_fill_form_success(self, client, mock_controller, db):
         assert data["output_pdf_path"] == "src/outputs/filled_output.pdf"
         mock_controller["form_ctrl"].fill_form.assert_called_once()
 
+        fetched = db.get(FormSubmission, data["id"])
+        assert fetched.input_id == input_id
+
     def test_fill_form_missing_template(self, client, mock_controller):
         resp = client.post(f"{API_PREFIX}/forms/fill", json={
             "template_id": 9999,
@@ -437,4 +449,5 @@ def test_full_flow(self, client, mock_controller, pdf_upload, tmp_path, monkeypa
         db_forms = list(db.exec(select(FormSubmission)))
         assert len(db_forms) == 1
         assert db_forms[0].template_id == template_id
+        assert db_forms[0].input_id == input_record.input_id
         assert "Jane Smith" in db_forms[0].input_text

tests/test_jobs.py

@@ -43,7 +43,7 @@ def test_submit_async_single(self, mock_task, client, db):
         assert data["jobs"][0]["status"] == "queued"
         assert "job_id" in data["jobs"][0]
         assert data["jobs"][0]["poll_url"].startswith(f"{API_PREFIX}/jobs/")
-        mock_task.delay.assert_called_once_with(tpl_id, "John Doe firefighter", None)
+        mock_task.delay.assert_called_once_with(tpl_id, "John Doe firefighter", str(input_id), None)
 
     @patch("app.api.routes.jobs.fill_form_task")
     def test_submit_async_batch(self, mock_task, client, db):
@@ -153,7 +153,7 @@ def test_submit_with_model_override(self, mock_task, client, db):
             "model": "mistral:latest",
         })
         assert resp.status_code == 200
-        mock_task.delay.assert_called_once_with(tpl_id, "test", "mistral:latest")
+        mock_task.delay.assert_called_once_with(tpl_id, "test", str(input_id), "mistral:latest")
 
     def test_submit_empty_template_ids(self, client, db):
         input_id = self._seed_input(db, transcript="test")

tests/test_migrations.py

@@ -79,17 +79,19 @@ def test_formsubmission_columns(alembic_cfg, alembic_engine):
 
     inspector = inspect(alembic_engine)
     columns = {c["name"] for c in inspector.get_columns("formsubmission")}
-    assert columns == {"id", "template_id", "input_text", "output_pdf_path", "created_at"}
+    assert columns == {
+        "id", "template_id", "input_id", "input_text", "output_pdf_path", "created_at"
+    }
 
 
 def test_formsubmission_fk(alembic_cfg, alembic_engine):
     command.upgrade(alembic_cfg, "head")
 
     inspector = inspect(alembic_engine)
-    fks = inspector.get_foreign_keys("formsubmission")
-    assert len(fks) == 1
-    assert fks[0]["referred_table"] == "template"
-    assert fks[0]["referred_columns"] == ["id"]
+    fks = {fk["referred_table"]: fk for fk in inspector.get_foreign_keys("formsubmission")}
+    assert len(fks) == 2
+    assert fks["template"]["referred_columns"] == ["id"]
+    assert fks["inputs"]["referred_columns"] == ["input_id"]
 
 
 def test_job_columns(alembic_cfg, alembic_engine):
@@ -293,9 +295,9 @@ def test_reports_no_fk(alembic_cfg, alembic_engine):
 
 
 def test_downgrade_002(alembic_cfg, alembic_engine):
-    """Downgrade by one step removes only the 002 tables, leaving 001 tables intact."""
+    """Downgrading to 001 removes the 002 tables, leaving 001 tables intact."""
     command.upgrade(alembic_cfg, "head")
-    command.downgrade(alembic_cfg, "-1")
+    command.downgrade(alembic_cfg, "001")
 
     inspector = inspect(alembic_engine)
     tables = inspector.get_table_names()
@@ -307,3 +309,16 @@ def test_downgrade_002(alembic_cfg, alembic_engine):
     assert "template" in tables
     assert "formsubmission" in tables
     assert "job" in tables
+
+
+def test_downgrade_003(alembic_cfg, alembic_engine):
+    """Downgrade by one step from head removes only the input_id FK/column."""
+    command.upgrade(alembic_cfg, "head")
+    command.downgrade(alembic_cfg, "-1")
+
+    inspector = inspect(alembic_engine)
+    columns = {c["name"] for c in inspector.get_columns("formsubmission")}
+    assert "input_id" not in columns
+    assert "input_text" in columns
+    tables = inspector.get_table_names()
+    assert "inputs" in tables