state-hub/api/routers/memo_notifications.py
tegwick 49eeddbfed
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Build and Publish Multi-Context Image / build-and-push (push) Successful in 27s
Add a dev memo-forwarding fixture and deduplicated intake notifications.
Pose, replay, and conflict stay on an injected transport with local retry
evidence. Notification receipts keep only memo id, question, class, and
submitter, and a resolve conflict stays readable.

Assistant: grok
Assistant-Session: 01a10329-31cb-78d3-b7a3-789e1fb83139
2026-10-03 21:25:46 +02:00

97 lines
3.4 KiB
Python

"""Notification intake for informed-decision delivery ids.
Intentionally absent from the 14-day write-idempotency allowlist: retention
is this receipt row, and a conflicting reuse must stay readable by GET.
"""
from __future__ import annotations
import json
from fastapi import APIRouter, Depends, Request
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
from api.database import get_session
from api.models.memo_notification import MemoNotificationReceipt
from api.services.informed_decision_contract import (
MAX_NOTIFICATION_BYTES,
ContractError,
parse_delivery_id,
parse_notification,
)
from api.services.memo_notification import NotificationConflict, accept_notification
router = APIRouter(prefix="/informed-decision", tags=["informed-decision-notifications"])
def _error(status: int, code: str) -> JSONResponse:
return JSONResponse(
status_code=status,
content={"error": code},
headers={"Cache-Control": "no-store"},
)
def _payload(receipt: MemoNotificationReceipt, *, status: int, replay: bool = False) -> JSONResponse:
return JSONResponse(
status_code=status,
content={
"delivery_id": receipt.delivery_id,
"memo_id": receipt.memo_id,
"question": receipt.question,
"memo_class": receipt.memo_class,
"submitter": receipt.submitter,
"message_id": str(receipt.message_id),
"status": "accepted",
"authoritative_review": receipt.authoritative_review,
},
headers={
"Cache-Control": "no-store",
**({"X-StateHub-Idempotency-Replay": "true"} if replay else {}),
},
)
@router.post("/notifications")
async def post_notification(
request: Request, session: AsyncSession = Depends(get_session)
) -> JSONResponse:
raw = await request.body()
if len(raw) > MAX_NOTIFICATION_BYTES:
return _error(413, "notification_too_large")
content_type = request.headers.get("content-type", "")
if content_type.split(";")[0].strip().lower() != "application/json":
return _error(415, "json_required")
try:
delivery_id = parse_delivery_id(request.headers.get("idempotency-key"))
notice = parse_notification(json.loads(raw))
except ContractError as exc:
return _error(exc.status, exc.code)
except (json.JSONDecodeError, UnicodeError):
return _error(400, "invalid_json")
try:
receipt, replay = await accept_notification(session, delivery_id, notice)
except NotificationConflict as exc:
return JSONResponse(
status_code=409,
content={
"error": "notification_conflict",
"delivery_id": exc.receipt.delivery_id,
},
headers={"Cache-Control": "no-store"},
)
return _payload(receipt, status=200 if replay else 201, replay=replay)
@router.get("/notifications/{delivery_id}")
async def get_notification(
delivery_id: str, session: AsyncSession = Depends(get_session)
) -> JSONResponse:
try:
checked = parse_delivery_id(delivery_id)
except ContractError as exc:
return _error(exc.status, exc.code)
receipt = await session.get(MemoNotificationReceipt, checked)
if receipt is None:
return _error(404, "notification_not_found")
return _payload(receipt, status=200)