98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import List, Optional
|
|
import uuid
|
|
|
|
from bd import Settings
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from bd.tables.poll import Poll
|
|
from bd.tables.question import Question
|
|
from bd.tables.response import Response, Answer
|
|
|
|
router = APIRouter(tags=["responses"], prefix="/polls")
|
|
|
|
|
|
class AnswerIn(BaseModel):
|
|
question_id: str
|
|
choice_id: Optional[str]
|
|
text: Optional[str]
|
|
|
|
|
|
class ResponseIn(BaseModel):
|
|
user_id: Optional[str]
|
|
answers: List[AnswerIn]
|
|
|
|
|
|
def get_session():
|
|
settings = Settings()
|
|
engine = create_engine(settings.DATABASE_URL_syncpg, future=True)
|
|
return sessionmaker(bind=engine, future=True)
|
|
|
|
|
|
@router.post("/{poll_id}/responses", status_code=201)
|
|
def submit_response(poll_id: str, payload: ResponseIn):
|
|
Session = get_session()
|
|
try:
|
|
pid = uuid.UUID(poll_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid poll UUID")
|
|
with Session() as session:
|
|
poll = session.get(Poll, pid)
|
|
if not poll:
|
|
raise HTTPException(status_code=404, detail="Poll not found")
|
|
resp = Response(poll_id=pid)
|
|
if payload.user_id:
|
|
try:
|
|
resp.user_id = uuid.UUID(payload.user_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid user_id UUID")
|
|
session.add(resp)
|
|
# validate answers
|
|
for a in payload.answers:
|
|
try:
|
|
qid = uuid.UUID(a.question_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid question_id UUID")
|
|
question = session.get(Question, qid)
|
|
if not question or question.poll_id != pid:
|
|
raise HTTPException(status_code=400, detail="Question does not belong to poll")
|
|
choice_uuid = None
|
|
if a.choice_id:
|
|
try:
|
|
choice_uuid = uuid.UUID(a.choice_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid choice_id UUID")
|
|
answer = Answer(response=resp, question_id=qid, choice_id=choice_uuid, text=a.text)
|
|
session.add(answer)
|
|
session.commit()
|
|
session.refresh(resp)
|
|
return {"id": str(resp.id)}
|
|
|
|
|
|
@router.get("/{poll_id}/responses")
|
|
def list_responses(poll_id: str):
|
|
Session = get_session()
|
|
try:
|
|
pid = uuid.UUID(poll_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid poll UUID")
|
|
with Session() as session:
|
|
rows = session.query(Response).filter(Response.poll_id == pid).all()
|
|
return [{"id": str(r.id), "user_id": str(r.user_id) if r.user_id else None, "submitted_at": r.submitted_at.isoformat()} for r in rows]
|
|
|
|
|
|
@router.get("/responses/{response_id}")
|
|
def get_response(response_id: str):
|
|
Session = get_session()
|
|
try:
|
|
rid = uuid.UUID(response_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid UUID")
|
|
with Session() as session:
|
|
r = session.get(Response, rid)
|
|
if not r:
|
|
raise HTTPException(status_code=404, detail="Response not found")
|
|
return {"id": str(r.id), "poll_id": str(r.poll_id), "user_id": str(r.user_id) if r.user_id else None, "submitted_at": r.submitted_at.isoformat(), "answers": [{"id": str(a.id), "question_id": str(a.question_id), "choice_id": str(a.choice_id) if a.choice_id else None, "text": a.text} for a in r.answers]}
|