Review
Admin CSV of Gantry damage claims for insurers
Gantry rents skid steers and boom lifts to job sites. Ops wants an admin endpoint that streams open damage claims as CSV so partner insurers can paste the file straight into Excel for adjusters — includes free-text description and witness_notes from the field app.
FastAPITier 2securitycsvinjection
Click a line to flag it, pick one or more labels, then submit. If the change looks correct, approve it.
services/admin/claims_export.py+28-0
| 1 | 1 | import csv | |
| 2 | 2 | import io | |
| 3 | 3 | ||
| 4 | 4 | from fastapi import APIRouter, Depends | |
| 5 | 5 | from fastapi.responses import StreamingResponse | |
| 6 | 6 | ||
| 7 | 7 | from app.auth import require_admin | |
| 8 | 8 | from app.db import claims_repo | |
| 9 | 9 | ||
| 10 | 10 | router = APIRouter(prefix="/admin/claims", tags=["admin"]) | |
| 11 | + | ||
| 12 | + | HEADERS = ["claim_id", "equipment_sku", "amount_cents", "description", "witness_notes", "status"] | |
| 13 | + | ||
| 14 | + | def _rows(claims): | |
| 15 | + | yield HEADERS | |
| 16 | + | for c in claims: | |
| 17 | + | yield [ | |
| 18 | + | c.id, | |
| 19 | + | c.equipment_sku, | |
| 20 | + | str(c.amount_cents), | |
| 21 | + | c.description, | |
| 22 | + | c.witness_notes, | |
| 23 | + | c.status, | |
| 24 | + | ] | |
| 25 | + | ||
| 26 | + | @router.get("/export.csv") | |
| 27 | + | async def export_open_claims(_: None = Depends(require_admin)): | |
| 28 | + | claims = await claims_repo.list_open() | |
| 29 | + | buf = io.StringIO() | |
| 30 | + | writer = csv.writer(buf) | |
| 31 | + | for row in _rows(claims): | |
| 32 | + | writer.writerow(row) | |
| 33 | + | buf.seek(0) | |
| 34 | + | return StreamingResponse( | |
| 35 | + | iter([buf.getvalue()]), | |
| 36 | + | media_type="text/csv", | |
| 37 | + | headers={"Content-Disposition": 'attachment; filename="gantry-claims.csv"'}, | |
| 38 | + | ) |