[verified] add exam-board signed URL endpoint
api-ci-deploy / test-build-deploy (push) Has been cancelled

(cherry picked from commit c65d18ca6b)
This commit is contained in:
CC Worker
2026-06-08 01:51:55 +00:00
parent c69451fba2
commit 34fc7edd68
2 changed files with 127 additions and 4 deletions
+65 -4
View File
@@ -137,6 +137,22 @@ def _lookup_exam_storage_loc(exam_id: str) -> Optional[str]:
return None
def _signed_url_value(result: Any) -> str:
"""Normalise supabase-py signed URL responses across v1/v2 shapes."""
if isinstance(result, str):
return result
if isinstance(result, dict):
value = result.get("signedURL") or result.get("signedUrl") or result.get("signed_url")
if value:
return str(value)
data = getattr(result, "data", None)
if isinstance(data, dict):
value = data.get("signedURL") or data.get("signedUrl") or data.get("signed_url")
if value:
return str(value)
raise ValueError("Storage service did not return a signed URL")
async def _parse_create_template_request(request: Request) -> tuple[CreateTemplateRequest, Optional[UploadFile]]:
content_type = request.headers.get("content-type", "")
if "multipart/form-data" in content_type:
@@ -608,12 +624,13 @@ async def create_template(
@router.get("/catalogue")
async def list_catalogue_papers() -> Dict[str, Any]:
"""Lightweight exam-board paper catalogue for the create dialog."""
async def list_catalogue_papers(
ctx: ExamContext = Depends(get_exam_context),
) -> Dict[str, Any]:
"""Lightweight authenticated exam-board metadata catalogue for the create dialog."""
try:
sb = SupabaseServiceRoleClient().supabase
res = (
sb.table("eb_exams")
ctx.supabase.table("eb_exams")
.select("id, exam_code, spec_code, paper_code, tier, session, type_code, storage_loc")
.eq("type_code", "QP")
.order("exam_code")
@@ -624,6 +641,50 @@ async def list_catalogue_papers() -> Dict[str, Any]:
raise HTTPException(status_code=502, detail=f"Could not load catalogue papers: {exc}")
@router.get("/catalogue/{exam_id}/signed-url")
async def get_catalogue_paper_signed_url(
exam_id: str,
expires_in: int = 300,
ctx: ExamContext = Depends(get_exam_context),
) -> Dict[str, Any]:
"""Return a short-lived signed URL for an authenticated user's catalogue PDF access.
The storage operation uses service role as a scoped backend exception for signing only;
raw cc.examboards object reads remain denied by storage.objects RLS.
"""
expires_in = max(60, min(int(expires_in or 300), 3600))
try:
row = _first(
ctx.supabase.table("eb_exams")
.select("id, exam_code, storage_loc")
.eq("id", exam_id)
.eq("type_code", "QP")
.limit(1)
.execute()
)
if not row or not row.get("storage_loc"):
raise HTTPException(status_code=404, detail="Catalogue paper not found")
try:
bucket, path = _parse_storage_loc(row["storage_loc"])
except ValueError:
raise HTTPException(status_code=404, detail="Catalogue paper not found")
if bucket != "cc.examboards":
raise HTTPException(status_code=404, detail="Catalogue paper not found")
signed_url = _signed_url_value(StorageAdmin().create_signed_url(bucket, path, expires_in))
return {
"exam_id": row["id"],
"exam_code": row.get("exam_code"),
"bucket": bucket,
"path": path,
"expires_in": expires_in,
"signed_url": signed_url,
}
except HTTPException:
raise
except Exception as exc:
raise HTTPException(status_code=502, detail=f"Could not sign catalogue paper URL: {exc}")
@router.get("/templates")
async def list_templates(
include_archived: bool = False,