Building a Bulk-Approve Exception Grid

A reviewer working a Batch Approval UI Patterns queue rarely wants to approve exceptions one row at a time. Once the Manual Review Queue accumulates enough near-duplicate items — same counterparty, same currency, same value date, amounts inside a tolerance band — the efficient move is to group them and approve the whole group in a single action. The client-side rendering of that grid is out of scope here; this page builds the server-side contract underneath it: a deterministic grouping key so the same exceptions land in the same group across page reloads, a keyset-paginated query the grid polls, dual-control enforcement so the same person cannot both stage and approve a batch, and an atomic bulk-approve handler that uses optimistic concurrency to guard against a stale page. Every commit — full or partial — writes an audit record carrying trace_id, source_hash, and match_decision, and an approved batch flows on to Automating Batch Reconciliation Sign-Offs once posted to the ledger.

Prerequisites

• Exceptions already carry a materiality_score and routing tier assigned upstream by Threshold-Based Routing Logic.
• A Manual Review Queue table backs the grid, with an integer version_token column for optimistic concurrency.
• Python 3.10+ with pydantic v2 and fastapi installed; monetary and materiality fields use Decimal, never float.
• A structured logger configured to emit trace_id, source_hash, and match_decision on every batch commit.
• Two distinct reviewer identities are available so dual-control enforcement can be exercised in a staging environment.

Step 1 — Group exceptions with a deterministic hash

The grouping key has to be stable across requests: if the grid re-fetches a page a second later, the same exceptions must land in the same group, or a reviewer’s in-flight selection silently orphans. Hashing a small set of normalised fields — never the raw, un-normalised strings, which vary in whitespace and casing between feed formats — gives that stability. Amounts are bucketed rather than compared exactly, since exceptions in the same logical group are rarely identical to the cent.

import hashlib
from decimal import Decimal
from datetime import date

def normalise_group_key(
    counterparty: str,
    currency: str,
    amount: Decimal,
    value_date: date,
    amount_bucket_usd: Decimal = Decimal("50.00"),
) -> str:
    """Deterministic group key: identical normalised inputs always hash the
    same way, so a re-fetched grid page never reshuffles a reviewer's groups."""
    normalised_counterparty = " ".join(counterparty.strip().upper().split())
    bucket = (amount // amount_bucket_usd) * amount_bucket_usd
    blob = "|".join(
        [
            normalised_counterparty,
            currency.strip().upper(),
            str(bucket.quantize(Decimal("0.01"))),
            value_date.isoformat(),
        ]
    )
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]

The 16-character truncation keeps the key compact enough to index and to embed in a URL query string when the grid deep-links to a specific batch, while the underlying SHA-256 keeps collisions negligible for any realistic exception volume. Persist group_key as a stored, indexed column on write rather than recomputing it on every read — recomputation is fine for verification, but the paged query in the next step needs to ORDER BY it directly.

Step 2 — Serve the grid as a paged, ordered query

A bulk-approve grid is read far more often than it is written, so the paging strategy matters more than the write path. Offset-based paging (LIMIT/OFFSET) degrades as the queue grows and produces duplicate or skipped rows when items change state between page loads. Keyset pagination on (group_key, exception_id) avoids both problems: it is a stable index scan regardless of table size, and it composes naturally with the deterministic grouping from Step 1, since rows in the same group always sort adjacently.

from uuid import UUID
from decimal import Decimal
from pydantic import BaseModel

class ExceptionRow(BaseModel):
    exception_id: UUID
    group_key: str
    counterparty: str
    currency: str
    amount: Decimal
    materiality_score: Decimal
    version_token: int
    source_hash: str

GRID_PAGE_SQL = """
SELECT exception_id, group_key, counterparty, currency, amount,
       materiality_score, version_token, source_hash
FROM   exception_grid
WHERE  state = 'PENDING_REVIEW'
  AND  (group_key, exception_id) > ($1, $2)
ORDER  BY group_key, exception_id
LIMIT  $3
"""

async def fetch_grid_page(
    conn, after_group_key: str, after_exception_id: UUID, page_size: int = 100
) -> list[ExceptionRow]:
    rows = await conn.fetch(GRID_PAGE_SQL, after_group_key, after_exception_id, page_size)
    return [ExceptionRow(**dict(r)) for r in rows]

Each row surfaces its own version_token, which the grid must round-trip back on approval. That token is what lets the commit handler in Step 4 detect that a row changed underneath a reviewer between the page load and the click — the mechanism a plain “select all, approve all” endpoint has no way to express.

Step 3 — Enforce dual control before anything commits

Financial controls frameworks generally require that the person who stages a batch cannot also be the sole approver of that same batch — a segregation-of-duties (SoD) rule. The check has to run before any row is touched, not as an afterthought on the response, or a caller that skips the UI and calls the API directly can bypass it entirely.

class SodViolation(Exception):
    """Raised when the same principal both proposed and approved a batch."""

def enforce_dual_control(
    proposer_id: str, approver_id: str, require_dual_control: bool
) -> None:
    if require_dual_control and proposer_id == approver_id:
        raise SodViolation(
            f"approver {approver_id!r} cannot approve their own batch"
        )

require_dual_control is deliberately a parameter rather than a hard-coded True: some deployments run it per routing tier, waiving the rule only for the lowest-materiality tier where a compensating control (post-hoc sampling) already exists. Treat that waiver as a configuration decision recorded in the boundary table below, not a code branch scattered through the handler.

Step 4 — Commit the batch atomically with optimistic concurrency

This is the handler the grid actually calls. It re-reads the current state of every row in the group under FOR UPDATE, rejects the whole batch if any version_token no longer matches what the reviewer’s page showed, rejects it if the group mixes materiality tiers above the configured cap, and only then applies every update inside a single transaction. If any single update fails partway through, the transaction rolls back rather than leaving some items APPROVED and others still PENDING_REVIEW — a bulk action is either all-or-nothing from the ledger’s point of view.

import logging
from datetime import datetime, timezone
from decimal import Decimal
from uuid import UUID

from fastapi import APIRouter
from pydantic import BaseModel

log = logging.getLogger("batch.approve")
router = APIRouter()

MATERIALITY_CAP = Decimal("25000.00")


class BulkApproveItem(BaseModel):
    exception_id: UUID
    expected_version_token: int


class BulkApproveRequest(BaseModel):
    group_key: str
    proposer_id: str
    approver_id: str
    items: list[BulkApproveItem]


class StaleBatchError(Exception):
    pass


class PartialCommitError(Exception):
    pass


class MixedMaterialityError(Exception):
    pass


@router.post("/exceptions/batches/{group_key}/approve")
async def bulk_approve(group_key: str, req: BulkApproveRequest, db) -> dict:
    enforce_dual_control(req.proposer_id, req.approver_id, require_dual_control=True)

    async with db.transaction() as tx:
        rows = await tx.fetch(
            "SELECT exception_id, version_token, materiality_score, source_hash "
            "FROM exception_grid WHERE group_key = $1 FOR UPDATE",
            group_key,
        )
        current = {r["exception_id"]: r for r in rows}

        if any(r["materiality_score"] > MATERIALITY_CAP for r in rows):
            raise MixedMaterialityError(
                f"group {group_key} contains an item above materiality_cap"
            )

        stale = [
            item.exception_id
            for item in req.items
            if item.exception_id not in current
            or current[item.exception_id]["version_token"] != item.expected_version_token
        ]
        if stale:
            raise StaleBatchError(f"stale version tokens in group {group_key}: {stale}")

        approved_at = datetime.now(timezone.utc)
        try:
            for item in req.items:
                row = current[item.exception_id]
                await tx.execute(
                    "UPDATE exception_grid "
                    "SET state = 'APPROVED', version_token = version_token + 1, "
                    "    approved_by = $2, approved_at = $3 "
                    "WHERE exception_id = $1 AND version_token = $4",
                    item.exception_id,
                    req.approver_id,
                    approved_at,
                    item.expected_version_token,
                )
                log.info(
                    "batch.item_approved",
                    extra={
                        "trace_id": str(item.exception_id),
                        "source_hash": row["source_hash"],
                        "match_decision": "BULK_APPROVED",
                        "group_key": group_key,
                        "approver_id": req.approver_id,
                    },
                )
        except Exception as exc:
            await tx.rollback()
            raise PartialCommitError(f"batch {group_key} rolled back: {exc}") from exc

    return {
        "group_key": group_key,
        "approved": len(req.items),
        "approved_at": approved_at.isoformat(),
    }

The WHERE exception_id = $1 AND version_token = $4 clause on the UPDATE is the actual optimistic-concurrency guard — even if the earlier SELECT ... FOR UPDATE momentarily locked stale data due to a race with a concurrent transaction, a mismatched token on the write itself still blocks the update. Map StaleBatchError, PartialCommitError, MixedMaterialityError, and SodViolation to distinct HTTP status codes (409, 500, 422, 403) in a FastAPI exception handler so the grid can render a specific retry action instead of a generic failure banner.

Configuration boundary table

Verification and testing

Because the handler’s correctness hinges on the transaction rolling back cleanly, test it against a fake connection that records every execute call and every rollback, rather than a live database — the assertion is about what the handler attempts, not about actual row state. A fixture with a deliberately mismatched version_token exercises the stale-batch path without needing a second concurrent writer.

import pytest
from uuid import uuid4
from decimal import Decimal

def test_stale_batch_blocks_commit():
    exception_id = uuid4()
    current = {
        exception_id: {
            "version_token": 3,
            "materiality_score": Decimal("500.00"),
            "source_hash": "a1b2c3",
        }
    }
    req = BulkApproveRequest(
        group_key="grp-001",
        proposer_id="alice",
        approver_id="bob",
        items=[BulkApproveItem(exception_id=exception_id, expected_version_token=2)],
    )
    stale = [
        item.exception_id
        for item in req.items
        if current[item.exception_id]["version_token"] != item.expected_version_token
    ]
    with pytest.raises(StaleBatchError):
        if stale:
            raise StaleBatchError(f"stale version tokens: {stale}")


def test_dual_control_rejects_self_approval():
    with pytest.raises(SodViolation):
        enforce_dual_control("alice", "alice", require_dual_control=True)

Run the stale-batch scenario as an integration test too, with two real transactions: open one connection, update a row’s version_token mid-test, then submit the original (now stale) token through bulk_approve and assert the response is a 409 with StaleBatchError in the body, not a silently successful partial commit.

Troubleshooting

• STALE_BATCH — the commit is rejected even though the reviewer just loaded the page. Root cause: another worker or reviewer mutated a row in the group after the page rendered but before the click. Fix: return the current version_token values in the 409 response body so the grid can refresh only the changed rows and let the reviewer resubmit without losing their other selections.
• PARTIAL_COMMIT — some items show APPROVED while others in the same batch stayed PENDING_REVIEW. Root cause: updates ran outside a single transaction, or an exception was swallowed without a rollback() call. Fix: keep every UPDATE inside one db.transaction() block as in Step 4, and never catch-and-continue inside the loop.
• MIXED_MATERIALITY — a batch is rejected that looked uniform in the grid. Root cause: the grid’s client-side filter and the server’s materiality_cap check disagree, usually because the client cached a stale materiality_score. Fix: treat the server check as authoritative and re-render the grid page after any rejection rather than trusting the client’s copy.
• SOD_VIOLATION — a legitimate reviewer cannot approve their own staged batch. Root cause: enforce_dual_control is working as designed — this is the segregation-of-duties control, not a bug. Fix: route the batch to a second approver instead of disabling the check; if a tier genuinely has only one reviewer, set require_dual_control to false for that tier explicitly and document the compensating control.

Related

• Automating Batch Reconciliation Sign-Offs
• Manual Review Queue Design
• Skip-Locked Queue Consumption in Postgres
• Calibrating Materiality Thresholds From History

Part of Batch Approval UI Patterns within Exception Routing & Human-in-the-Loop Workflows.