feat(review): page, decide and edit stored upload rows, insert-only (M4a, #108)

alembic/versions/0008_review.py

@@ -0,0 +1,69 @@
+"""Review of stored uploads: row decisions and versioned cell edits (M4a, #108; BR-25, BR-26, BR-27).
+
+Decided 28 Oct: review is recorded now and applied by M6's staged import; the
+imported table doesn't change. A reason is required.
+
+- ``uploads`` gains ``columns`` and ``column_types``: the sanitised names and
+  types the upload used, so every stored row can be validated.
+- ``row_decisions``: include or exclude a row, with actor, reason and time.
+- ``cell_edits``: one row per edit, with old and new value, actor, reason and
+  time. The original stays in ``source_rows``.
+
+Both new tables are insert-only (the latest entry wins; earlier ones are the
+history), with the same guard as 0006 and 0007. Runs per tenant.
+
+Revision ID: 0008
+Revises: 0007
+Create Date: 2026-10-28
+"""
+
+from alembic import op
+
+from api.config import settings
+
+revision = "0008"
+down_revision = "0007"
+branch_labels = None
+depends_on = None
+
+_TABLES = ("row_decisions", "cell_edits")
+
+
+def _schema() -> str:
+    return op.get_bind().dialect.identifier_preparer.quote(settings.UPLOADS_SCHEMA)
+
+
+def upgrade() -> None:
+    s = _schema()
+    op.execute(f"ALTER TABLE {s}.uploads ADD COLUMN columns TEXT[], ADD COLUMN column_types TEXT[]")
+    op.execute(f"""CREATE TABLE {s}.row_decisions (
+        id BIGSERIAL PRIMARY KEY,
+        upload_id BIGINT NOT NULL REFERENCES {s}.uploads (id),
+        row_number INTEGER NOT NULL CHECK (row_number > 0),
+        decision TEXT NOT NULL CHECK (decision IN ('include', 'exclude')),
+        reason TEXT NOT NULL CHECK (btrim(reason) <> ''),
+        actor TEXT,
+        decided_at TIMESTAMPTZ NOT NULL DEFAULT now())""")
+    op.execute(f"""CREATE TABLE {s}.cell_edits (
+        id BIGSERIAL PRIMARY KEY,
+        upload_id BIGINT NOT NULL REFERENCES {s}.uploads (id),
+        row_number INTEGER NOT NULL CHECK (row_number > 0),
+        column_index INTEGER NOT NULL CHECK (column_index >= 0),
+        old_value TEXT NOT NULL,
+        new_value TEXT NOT NULL,
+        reason TEXT NOT NULL CHECK (btrim(reason) <> ''),
+        actor TEXT,
+        edited_at TIMESTAMPTZ NOT NULL DEFAULT now())""")
+    for t in _TABLES:
+        op.execute(f"CREATE INDEX {t}_by_row ON {s}.{t} (upload_id, row_number, id)")
+        op.execute(f"CREATE TRIGGER {t}_no_change BEFORE UPDATE OR DELETE ON {s}.{t} "
+                   f"FOR EACH ROW EXECUTE FUNCTION {s}.audit_reject_change()")
+        op.execute(f"CREATE TRIGGER {t}_no_truncate BEFORE TRUNCATE ON {s}.{t} "
+                   f"FOR EACH STATEMENT EXECUTE FUNCTION {s}.audit_reject_change()")
+
+
+def downgrade() -> None:
+    s = _schema()
+    for t in reversed(_TABLES):
+        op.execute(f"DROP TABLE IF EXISTS {s}.{t}")
+    op.execute(f"ALTER TABLE {s}.uploads DROP COLUMN columns, DROP COLUMN column_types")

api/auth.py

@@ -88,6 +88,7 @@ async def tenant_scope(authorization: str = Header(default=""),
     "delete_file": frozenset({"tenant_admin"}),
     "read_audit":  frozenset({"tenant_admin", "auditor"}),
     "reset_password": frozenset({"tenant_admin"}),
+    "review_rows": frozenset({"reviewer", "tenant_admin"}),
 }
 
 

api/db.py

@@ -202,7 +202,8 @@ def owner_connect():
 
 
 # Tables the runtime role may only read and insert (M3, #107). Triggers also stop the owner.
-APPEND_ONLY = ("audit_log", "uploads", "upload_blobs", "source_rows", "alembic_version")
+APPEND_ONLY = ("audit_log", "uploads", "upload_blobs", "source_rows", "row_decisions", "cell_edits",
+               "alembic_version")
 
 
 def grant_runtime(cur, schema: str, create: bool) -> None:

api/main.py

@@ -18,7 +18,7 @@
 from api import db, oidc
 from api.config import settings
 from api.ratelimit import rate_limit_middleware
-from api.routers import audit_routes, auth_routes, csv_routes, te_routes
+from api.routers import audit_routes, auth_routes, csv_routes, review_routes, te_routes
 
 logger = logging.getLogger(__name__)
 
@@ -86,7 +86,7 @@ async def lifespan(app: FastAPI):
 app.add_middleware(
     CORSMiddleware,
     allow_origins=settings.CORS_ORIGINS,
-    allow_methods=["GET", "POST", "DELETE"],
+    allow_methods=["GET", "POST", "PUT", "DELETE"],  # PUT: review changes (M4, #108)
     allow_headers=["Content-Type", "X-API-Key", "Authorization"],
 )
 
@@ -97,6 +97,7 @@ async def lifespan(app: FastAPI):
 
 app.include_router(auth_routes.router)
 app.include_router(csv_routes.router)
+app.include_router(review_routes.router)
 app.include_router(te_routes.router)
 app.include_router(audit_routes.router)
 

api/routers/review_routes.py

@@ -0,0 +1,56 @@
+"""Review endpoints for stored uploads (M4a, #108; BR-25, BR-26, BR-27). See api/services/review.py."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from api.auth import allow, tenant_scope
+from api.db import Conn
+from api.services import review
+
+router = APIRouter(prefix="/api/csv/uploads", tags=["review"], dependencies=[Depends(tenant_scope)])
+
+
+class DecisionRequest(BaseModel):
+    decision: str
+    reason: str
+
+
+class EditRequest(BaseModel):
+    value: str
+    reason: str
+
+
+def _run(fn, *args, commit: bool = False):
+    with Conn() as conn:
+        with conn.cursor() as cur:
+            try:
+                result = fn(cur, *args)
+            except review.ReviewError as exc:
+                conn.rollback()
+                raise HTTPException(exc.status, str(exc)) from None
+        if commit:
+            conn.commit()
+    return result
+
+
+@router.get("/{upload_id}/rows", dependencies=[Depends(allow("read_rows"))])
+def rows(upload_id: int, offset: int = 0, limit: int = 100) -> dict:
+    """One page of every stored row, with current values, validation errors and decisions."""
+    return _run(review.page, upload_id, offset, limit)
+
+
+@router.put("/{upload_id}/rows/{row_number}/decision", dependencies=[Depends(allow("review_rows"))])
+def decide(upload_id: int, row_number: int, req: DecisionRequest) -> dict:
+    return _run(review.decide, upload_id, row_number, req.decision, req.reason, commit=True)
+
+
+@router.put("/{upload_id}/rows/{row_number}/cells/{column}", dependencies=[Depends(allow("review_rows"))])
+def edit(upload_id: int, row_number: int, column: str, req: EditRequest) -> dict:
+    return _run(review.edit, upload_id, row_number, column, req.value, req.reason, commit=True)
+
+
+@router.get("/{upload_id}/rows/{row_number}/history", dependencies=[Depends(allow("read_rows"))])
+def history(upload_id: int, row_number: int) -> list[dict]:
+    return _run(review.history, upload_id, row_number)

api/services/audit.py

@@ -38,7 +38,7 @@
 
 # Actions the log is allowed to record. A typo becomes a loud failure here rather
 # than an unqueryable string in the table.
-VALID_ACTIONS = frozenset({"upload", "delete", "password_reset"})
+VALID_ACTIONS = frozenset({"upload", "delete", "password_reset", "row_decision", "cell_edit"})
 
 # Who is acting (M3a, #107). tenant_scope sets it per request: the signed-in
 # username, or "api_key" for a request made with the shared key.

api/services/dynamic_loader.py

@@ -330,7 +330,8 @@ def _do_upload(
                     msg += f' First problem: row {e["rowNumber"]}, column "{e["column"]}": {e["reason"]}'
                 _log(logs, "error", msg, "error")
                 return {"status": "error", "message": msg, "rowErrors": row_errors,
-                        "totalRows": len(data_rows), "failedRows": failed_row_count, "logs": logs}
+                        "totalRows": len(data_rows), "failedRows": failed_row_count,
+                        "columns": columns, "types": col_types, "logs": logs}
 
             # Insert
             inserted = 0

api/services/originals.py

@@ -46,13 +46,13 @@ def record_upload(file_name: str, content: str, mode: str, result: dict) -> int
                 s = sql.Identifier(uploads_schema())
                 cur.execute(sql.SQL(
                     "INSERT INTO {}.uploads (file_name, sha256, byte_size, uploader, mode, status, "
-                    "total_rows, imported_rows, failed_rows, csv_file_id, row_errors, logs) "
-                    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id").format(s),
+                    "total_rows, imported_rows, failed_rows, csv_file_id, row_errors, logs, columns, column_types) "
+                    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id").format(s),
                     (file_name, hashlib.sha256(data).hexdigest(), len(data), current_actor(), mode,
                      outcome(result), result.get("totalRows"), result.get("insertedRows"),
                      result.get("failedRows"), coerce_file_id(result.get("fileId")) if ok else None,
                      Json((result.get("rowErrors") or [])[:settings.MAX_ROW_ERRORS_REPORTED]),
-                     Json(result.get("logs") or [])))
+                     Json(result.get("logs") or []), result.get("columns"), result.get("types")))
                 upload_id = cur.fetchone()[0]
                 cur.execute(sql.SQL("INSERT INTO {}.upload_blobs (upload_id, content) VALUES (%s, %s)").format(s),
                             (upload_id, data))

api/services/review.py

@@ -0,0 +1,188 @@
+"""Review of a stored upload: every row, its validation, row decisions and cell edits (M4a, #108).
+
+Works on M3's ``source_rows``; the original is never changed. Decisions and
+edits are insert-only rows (migration 0008), and the latest one wins. They are
+applied by M6's staged import, not to the table imported at upload time
+(product owner, 28 Oct).
+"""
+
+from __future__ import annotations
+
+from psycopg2 import sql
+
+from api.services.audit import current_actor, write_audit
+from api.services.csv_parse import cast_value, extra_cells, ragged_row_reason, sanitize_columns
+from api.tenancy import uploads_schema
+
+MAX_PAGE = 500
+
+
+class ReviewError(ValueError):
+    """A request the API answers with 4xx: ``status`` is the HTTP code."""
+
+    def __init__(self, status: int, message: str):
+        super().__init__(message)
+        self.status = status
+
+
+def _s():
+    return sql.Identifier(uploads_schema())
+
+
+def _upload(cur, upload_id: int) -> dict:
+    cur.execute(sql.SQL(
+        "SELECT u.file_name, u.columns, u.column_types, h.cells, "
+        "  (SELECT count(*) FROM {s}.source_rows r WHERE r.upload_id = u.id AND r.row_number > 0) "
+        "FROM {s}.uploads u LEFT JOIN {s}.source_rows h ON h.upload_id = u.id AND h.row_number = 0 "
+        "WHERE u.id = %s").format(s=_s()), (upload_id,))
+    row = cur.fetchone()
+    if row is None:
+        raise ReviewError(404, "No such upload")
+    name, columns, types, header, total = row
+    columns = columns or sanitize_columns(header or [])
+    # Uploads stored before 0008, or in T&E mode, have no types: every value counts as text.
+    types = types if types and len(types) == len(columns) else ["text"] * len(columns)
+    return {"fileName": name, "columns": columns, "types": types, "total": total}
+
+
+def _validate(cells: list[str], raw: list[str], columns: list[str], types: list[str]) -> list[dict]:
+    errors = []
+    if extra_cells(raw, len(columns)):
+        errors.append({"column": "(beyond header)", "value": extra_cells(raw, len(columns))[0],
+                       "reason": ragged_row_reason(raw, len(columns))})
+    for name, value, col_type in zip(columns, cells, types):
+        ok, _, reason = cast_value(value, col_type)
+        if not ok:
+            errors.append({"column": name, "value": value, "reason": reason})
+    return errors
+
+
+def _latest(cur, upload_id: int, numbers: list[int]) -> tuple[dict, dict]:
+    """The current decision per row and the current value per edited cell."""
+    cur.execute(sql.SQL(
+        "SELECT DISTINCT ON (row_number) row_number, decision, reason, actor FROM {}.row_decisions "
+        "WHERE upload_id = %s AND row_number = ANY(%s) ORDER BY row_number, id DESC").format(_s()),
+        (upload_id, numbers))
+    decisions = {n: {"decision": d, "reason": r, "actor": a} for n, d, r, a in cur.fetchall()}
+    cur.execute(sql.SQL(
+        "SELECT DISTINCT ON (row_number, column_index) row_number, column_index, new_value "
+        "FROM {}.cell_edits WHERE upload_id = %s AND row_number = ANY(%s) "
+        "ORDER BY row_number, column_index, id DESC").format(_s()), (upload_id, numbers))
+    edits: dict = {}
+    for n, c, v in cur.fetchall():
+        edits.setdefault(n, {})[c] = v
+    return decisions, edits
+
+
+def _rows(cur, upload_id: int, meta: dict, numbers: list[int] | None = None,
+          offset: int = 0, limit: int = 100) -> list[dict]:
+    if numbers is None:
+        cur.execute(sql.SQL(
+            "SELECT row_number, cells FROM {}.source_rows WHERE upload_id = %s AND row_number > 0 "
+            "ORDER BY row_number OFFSET %s LIMIT %s").format(_s()), (upload_id, offset, limit))
+    else:
+        cur.execute(sql.SQL(
+            "SELECT row_number, cells FROM {}.source_rows WHERE upload_id = %s AND row_number = ANY(%s) "
+            "ORDER BY row_number").format(_s()), (upload_id, numbers))
+    source = cur.fetchall()
+    decisions, edits = _latest(cur, upload_id, [n for n, _ in source])
+    width = len(meta["columns"])
+    out = []
+    for n, raw in source:
+        original = (list(raw) + [""] * width)[:width]
+        current = [edits.get(n, {}).get(c, v) for c, v in enumerate(original)]
+        decision = decisions.get(n)
+        out.append({
+            "rowNumber": n, "original": original, "current": current,
+            "editedColumns": sorted(edits.get(n, {})),
+            "decision": decision["decision"] if decision else "include",
+            "decisionReason": decision["reason"] if decision else None,
+            "errors": _validate(current, raw, meta["columns"], meta["types"]),
+        })
+    return out
+
+
+def page(cur, upload_id: int, offset: int, limit: int) -> dict:
+    meta = _upload(cur, upload_id)
+    offset, limit = max(0, offset), max(1, min(limit, MAX_PAGE))
+    cur.execute(sql.SQL(
+        "SELECT (SELECT count(*) FROM (SELECT DISTINCT ON (row_number) decision FROM {s}.row_decisions "
+        "        WHERE upload_id = %s ORDER BY row_number, id DESC) d WHERE decision = 'exclude'), "
+        "       (SELECT count(DISTINCT row_number) FROM {s}.cell_edits WHERE upload_id = %s)").format(s=_s()),
+        (upload_id, upload_id))
+    excluded, edited = cur.fetchone()
+    return {**meta, "offset": offset, "limit": limit, "excludedRows": excluded, "editedRows": edited,
+            "rows": _rows(cur, upload_id, meta, offset=offset, limit=limit)}
+
+
+def _check_row(meta: dict, row_number: int) -> None:
+    if not 1 <= row_number <= meta["total"]:
+        raise ReviewError(404, f"Row {row_number} doesn't exist; this upload has {meta['total']} rows")
+
+
+def _reason(reason: str) -> str:
+    if not (reason or "").strip():
+        raise ReviewError(422, "A reason is required")
+    return reason.strip()
+
+
+def decide(cur, upload_id: int, row_number: int, decision: str, reason: str) -> dict:
+    """Include or exclude one row. The caller commits."""
+    meta = _upload(cur, upload_id)
+    _check_row(meta, row_number)
+    if decision not in ("include", "exclude"):
+        raise ReviewError(422, "decision must be 'include' or 'exclude'")
+    reason = _reason(reason)
+    [before] = _rows(cur, upload_id, meta, numbers=[row_number])
+    cur.execute(sql.SQL("INSERT INTO {}.row_decisions (upload_id, row_number, decision, reason, actor) "
+                        "VALUES (%s, %s, %s, %s, %s)").format(_s()),
+                (upload_id, row_number, decision, reason, current_actor()))
+    write_audit(cur, "row_decision", None, meta["fileName"], None, None,
+                before={"upload": upload_id, "row": row_number, "decision": before["decision"]},
+                after={"upload": upload_id, "row": row_number, "decision": decision, "reason": reason})
+    [after] = _rows(cur, upload_id, meta, numbers=[row_number])
+    return after
+
+
+def edit(cur, upload_id: int, row_number: int, column: str, value: str, reason: str) -> dict:
+    """Set one cell's value as a new version. ``column`` is a name or a 0-based index. The caller commits."""
+    meta = _upload(cur, upload_id)
+    _check_row(meta, row_number)
+    names = meta["columns"]
+    index = int(column) if column.isdigit() else (names.index(column) if column in names else -1)
+    if not 0 <= index < len(names):
+        raise ReviewError(404, f"No column {column!r}")
+    reason = _reason(reason)
+    [before] = _rows(cur, upload_id, meta, numbers=[row_number])
+    old = before["current"][index]
+    if value == old:
+        raise ReviewError(422, "The new value is the same as the current one")
+    cur.execute(sql.SQL("INSERT INTO {}.cell_edits (upload_id, row_number, column_index, old_value, new_value, "
+                        "reason, actor) VALUES (%s, %s, %s, %s, %s, %s, %s)").format(_s()),
+                (upload_id, row_number, index, old, value, reason, current_actor()))
+    cell = {"upload": upload_id, "row": row_number, "column": names[index]}
+    write_audit(cur, "cell_edit", None, meta["fileName"], None, None,
+                before={**cell, "value": old}, after={**cell, "value": value, "reason": reason})
+    [after] = _rows(cur, upload_id, meta, numbers=[row_number])
+    return after
+
+
+def history(cur, upload_id: int, row_number: int) -> list[dict]:
+    """Every decision and edit on one row, oldest first."""
+    meta = _upload(cur, upload_id)
+    _check_row(meta, row_number)
+    cur.execute(sql.SQL(
+        "SELECT 'decision', decided_at, actor, reason, decision, NULL::int, NULL, NULL FROM {s}.row_decisions "
+        "  WHERE upload_id = %s AND row_number = %s "
+        "UNION ALL SELECT 'edit', edited_at, actor, reason, NULL, column_index, old_value, new_value "
+        "  FROM {s}.cell_edits WHERE upload_id = %s AND row_number = %s ORDER BY 2").format(s=_s()),
+        (upload_id, row_number, upload_id, row_number))
+    out = []
+    for kind, at, actor, reason, decision, index, old, new in cur.fetchall():
+        entry = {"kind": kind, "at": at.isoformat(), "actor": actor, "reason": reason}
+        if kind == "decision":
+            entry["decision"] = decision
+        else:
+            entry.update(column=meta["columns"][index], oldValue=old, newValue=new)
+        out.append(entry)
+    return out