Letting an LLM write to your database safely: previews, idempotency keys and ETags in a small FastAPI API

class Problem(Exception):
    def __init__(self, status: int, error_type: str, detail: str, headers: Optional[dict] = None, **extra):
        self.status, self.error_type, self.detail = status, error_type, detail
        self.headers, self.extra = headers or {}, extra

def problem_response(p: Problem) -> JSONResponse:
    body = {"type": f"urn:nutriscan:error:{p.error_type}", "title": HTTPStatus(p.status).phrase,
            "status": p.status, "detail": p.detail, "error_type": p.error_type, **p.extra}
    return JSONResponse(jsonable_encoder(body), status_code=p.status, headers=p.headers,
                        media_type="application/problem+json")

if request.method in ("POST", "PATCH", "PUT", "DELETE"):
    declared = request.headers.get("content-length")
    if declared is None and request.headers.get("transfer-encoding"):
        # a chunked body would be read whole before its size is known
        raise Problem(411, "length_required", "Send a Content-Length header.")
    if declared and declared.isdigit() and int(declared) > m.APP_BODY_CAP:
        raise Problem(413, "payload_too_large", "Body over 64 KB.")
    body = await request.body()
    if len(body) > m.APP_BODY_CAP:
        raise Problem(413, "payload_too_large", "Body over 64 KB.")
    ctype = request.headers.get("content-type", "").split(";")[0].strip().lower()
    if body and ctype != "application/json":
        raise Problem(415, "unsupported_media_type", "Send JSON with Content-Type: application/json.")

Name = Annotated[str, Field(min_length=1, max_length=80), AfterValidator(_text)]
Servings = Annotated[float, Field(gt=0, le=100)]
LogDate = Annotated[str, Field(pattern=r"^\d{4}-\d{2}-\d{2}$"), AfterValidator(_log_date)]

class Strict(BaseModel):
    model_config = ConfigDict(extra="forbid")

class Macros(Strict):
    calories: float = Field(ge=0, le=MAX_KCAL)
    protein_g: float = Field(ge=0, le=1000)
    # ...
    sodium_mg: Optional[float] = Field(None, ge=0, le=100000)

class Item(Strict):
    name: Name
    portion: Optional[Portion] = None
    servings: Servings = 1
    macros: Macros
    save_to_library: bool = False

TOKEN_PREFIX = "nsk_live_"
_TOKEN_RE = re.compile(r"^nsk_live_[A-Za-z0-9_-]{43}$")
MAX_TOKENS = 10
EXPIRY_DAYS = {"30d": 30, "90d": 90, "1y": 365, "never": None}

def token_hash(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()

if token.startswith(TOKEN_PREFIX):
    known = _live_prefixes is None or token[:12] in _live_prefixes
    row = _TOKEN_RE.match(token) and known and _lookup_token(token_hash(token), request.headers.get("user-agent"))

if not row:
    ip = m._client_ip(request)
    if m._spike(f"patfail:{ip}", 20, 300):
        m._blocked[ip] = time.time() + 600
    raise Problem(401, "unauthorized", "Invalid token.")

def need(*scopes: str, write: bool = False):
    def dep(request: Request) -> Caller:
        budget_gate()
        caller = resolve_caller(request)
        _require(caller, *scopes)
        request.state.ratelimit = take_request(caller, write)
        # ...

BURST_PER_MIN, WRITES_PER_MIN, DAILY_CAP = 20, 10, 200
MELBOURNE = ZoneInfo("Australia/Melbourne")
# ...
retry = max(1, int(resets.timestamp() - now) + 1)
raise Problem(429, "rate_limited", f"Over the limit of {limit}. Try again after {_zulu(resets)}.",
              headers={"Retry-After": str(retry), "RateLimit-Limit": str(limit), "RateLimit-Remaining": "0"},
              limit=limit, used=used, resets_at=_zulu(resets))

@contextmanager
def db(user_id: Optional[str], commit: bool = True):
    conn = None
    try:
        conn = m.get_db(user_id)
        cur = conn.cursor()
        yield cur
        if commit:
            conn.commit()
        cur.close()
    # ...

with db(caller.user_id, commit=not preview) as cur:
    ...  # the exact same writes, triggers, RLS policies and validation

if not preview and request.method == "POST":
    key = (idempotency_key or "").strip()
    if not key or len(key) > 255:
        raise Problem(422, "validation_error", "Idempotency-Key header required (1-255 characters) on a real write.")
rhash = _request_hash(request.url.path, payload)
with db(caller.user_id, commit=not preview) as cur:
    if key:
        cur.execute("DELETE FROM api_idempotency WHERE user_id = %s AND key = %s "
                    "AND created_at < now() - interval '24 hours'", ...)
        cur.execute("""INSERT INTO api_idempotency (user_id, key, request_hash, status_code, response)
                       VALUES (%s, %s, %s, 0, '{}') ON CONFLICT (user_id, key) DO NOTHING RETURNING 1""", ...)
        if not cur.fetchone():
            old = ...  # the stored row
            if old[0] != rhash:
                raise Problem(409, "idempotency_conflict", "This Idempotency-Key was used for a different request.")
            if not old[1]:
                raise Problem(409, "idempotency_conflict", "A request with this Idempotency-Key is still running.",
                              in_progress=True)
            replay = (old[1], old[2])

def test_same_key_twice_gives_one_set_of_rows(self):
    body = {"date": TODAY, "name": "Apple", "macros": MACROS}
    first = self.post("/v1/entries", body)
    second = self.post("/v1/entries", body)
    self.assertEqual(second.json()["entry"]["log_id"], first.json()["entry"]["log_id"])
    self.assertEqual(len(self.statements("INSERT INTO daily_log")), 1)

def etag_of(*parts) -> str:
    return '"' + hashlib.sha256(json.dumps(parts, sort_keys=True, default=str).encode()).hexdigest()[:16] + '"'

def update_entry(self, c: UpdateEntryChange):
    log_id, name, servings, nutrition, d = self.entry(c.log_id)
    n = log_service.load_nutrition(nutrition)
    if etag_of(name, servings, n, str(d)) != c.if_match:
        raise Problem(412, "precondition_failed", "The entry changed since you read it. Read it again.")

def _if_match(value: Optional[str]) -> str:
    if not value:
        raise Problem(422, "validation_error", "If-Match header required: send the ETag you read.",
                      errors=[{"field": "If-Match", "message": "required"}])
    # Cloudflare weakens ETag to W/"..." when it compresses a response
    return value.strip().removeprefix("W/")

def budget_gate():
    """At 90% of the estimated free Neon month, /v1 pauses; the app keeps working."""
    if m.budget_used() >= 0.9:
        resume = m.neon_next_period_start(datetime.now(timezone.utc).date())
        retry = int(datetime(resume.year, resume.month, resume.day, tzinfo=timezone.utc).timestamp() - time.time())
        raise Problem(503, "api_paused_budget", f"The API is paused to protect the database budget until {resume.isoformat()}.",
                      headers={"Retry-After": str(max(retry, 1))}, resets_at=f"{resume.isoformat()}T00:00:00Z")

@mcp_router.get("/.well-known/oauth-protected-resource/mcp")
@mcp_router.get("/.well-known/oauth-protected-resource")
def protected_resource() -> dict:
    return {"resource": MCP_URL, "authorization_servers": [mcp_issuer()], "scopes_supported": ["email"],
            "bearer_methods_supported": ["header"], "resource_name": "NutriScan"}

def _mcp_challenge() -> Problem:
    return Problem(401, "unauthorized", "Connect through claude.ai to use NutriScan.",
                   headers={"WWW-Authenticate": f'Bearer resource_metadata="{MCP_PRM_URL}"'})

@mcp_router.api_route("/mcp", methods=["GET", "POST", "DELETE"])
async def mcp(request: Request):
    # parsed only after the origin and auth checks: a declared JSON body would 422 before them
    raw = await request.body()
    return await run_in_threadpool(_mcp, request, raw)

def _mcp(request: Request, raw: bytes):
    origin = request.headers.get("origin")
    if origin and origin not in MCP_ORIGINS:   # ("https://claude.ai", "https://claude.com")
        raise Problem(403, "forbidden_origin", "This origin may not call /mcp.")
    caller, client_id = mcp_caller(request)
    # ... only now: payload = json.loads(raw)

PENDING_TTL, PENDING_MAX = 600, 20

_, body = run_changes(request, caller, [change], True, None)
code, now = secrets.token_urlsafe(16), time.time()
with _pending_lock:
    for k in [k for k, v in _pending.items() if now - v[3] > PENDING_TTL]:
        del _pending[k]
    mine = sorted((v[3], k) for k, v in _pending.items() if v[0] == caller.user_id)
    for _, k in mine[:max(0, len(mine) - PENDING_MAX + 1)]:
        del _pending[k]
    _pending[code] = (caller.user_id, client_id, [change], now, guard)
return json.dumps({"preview": jsonable_encoder(body), "confirm_code": code,
                   "expires_in_minutes": 10, "next": NEXT_STEP}, default=str)

entry = _pending.get(code) if isinstance(code, str) else None
if (not entry or entry[0] != caller.user_id or entry[1] != client_id
        or time.time() - entry[3] > PENDING_TTL):
    raise Problem(422, "confirmation_invalid", "This confirmation expired or is not valid. Ask again.")
if entry[4] and _template_etag(caller, entry[2][0].template_id) != entry[4]:
    raise Problem(412, "precondition_failed", "This changed since the preview. Ask again.")
_, body = run_changes(request, caller, entry[2], False, code)   # the code is the idempotency key

def test_retry_replays_instead_of_saving_twice(self):
    code = self.preview()
    self.conn.executed.clear()   # the preview ran (and rolled back) its own INSERT
    first = self.confirm(code)
    second = self.confirm(code)   # e.g. the first answer was lost and claude.ai retried
    self.assertEqual(json.loads(second[1]), json.loads(first[1]))
    self.assertEqual(len(self.statements("INSERT INTO daily_log")), 1)

def _annotations(name: str) -> dict:
    saves = name == "confirm_change"
    return {"readOnlyHint": not saves, "destructiveHint": saves, "idempotentHint": False, "openWorldHint": False}

def test_published_openapi_matches_the_code(self):
    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "openapi-v1.json")
    with open(path, encoding="utf-8") as f:
        published = json.load(f)
    self.assertEqual(published, json.loads(json.dumps(api_v1.openapi_v1())),
                     "regenerate backend/openapi-v1.json from api_v1.openapi_v1()")
    routes = sum(len(ops) for ops in published["paths"].values())
    self.assertEqual(routes, 17)