117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import uuid
|
|
|
|
from bd import make_engine
|
|
|
|
from bd.tables.question import Question
|
|
from bd.tables.poll import Poll
|
|
from route.auth_utils import require_admin_key
|
|
|
|
router = APIRouter(tags=["questions"], prefix="/questions")
|
|
|
|
|
|
class QuestionCreate(BaseModel):
|
|
poll_id: str
|
|
text: str
|
|
type: Optional[str] = "single"
|
|
position: Optional[int]
|
|
|
|
|
|
class QuestionUpdate(BaseModel):
|
|
text: Optional[str]
|
|
type: Optional[str]
|
|
position: Optional[int]
|
|
|
|
|
|
def get_session():
|
|
from sqlalchemy.orm import sessionmaker
|
|
return sessionmaker(bind=make_engine(), future=True)
|
|
|
|
|
|
@router.post("/", status_code=201, dependencies=[Depends(require_admin_key)])
|
|
def create_question(payload: QuestionCreate):
|
|
Session = get_session()
|
|
try:
|
|
pid = uuid.UUID(payload.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")
|
|
q = Question(poll_id=pid, text=payload.text, type=payload.type or "single", position=payload.position)
|
|
session.add(q)
|
|
session.commit()
|
|
session.refresh(q)
|
|
return {"id": str(q.id)}
|
|
|
|
|
|
@router.put("/{question_id}", dependencies=[Depends(require_admin_key)])
|
|
def update_question(question_id: str, payload: QuestionUpdate):
|
|
Session = get_session()
|
|
try:
|
|
qid = uuid.UUID(question_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid UUID")
|
|
with Session() as session:
|
|
q = session.get(Question, qid)
|
|
if not q:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
if payload.text is not None:
|
|
q.text = payload.text
|
|
if payload.type is not None:
|
|
q.type = payload.type
|
|
if payload.position is not None:
|
|
q.position = payload.position
|
|
session.add(q)
|
|
session.commit()
|
|
session.refresh(q)
|
|
return {"id": str(q.id)}
|
|
|
|
|
|
@router.delete("/{question_id}", status_code=204, dependencies=[Depends(require_admin_key)])
|
|
def delete_question(question_id: str):
|
|
Session = get_session()
|
|
try:
|
|
qid = uuid.UUID(question_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid UUID")
|
|
with Session() as session:
|
|
q = session.get(Question, qid)
|
|
if not q:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
session.delete(q)
|
|
session.commit()
|
|
return {}
|
|
|
|
|
|
@router.get("/")
|
|
def list_questions(poll_id: Optional[str] = None):
|
|
Session = get_session()
|
|
with Session() as session:
|
|
q = session.query(Question)
|
|
if poll_id:
|
|
try:
|
|
pid = uuid.UUID(poll_id)
|
|
q = q.filter(Question.poll_id == pid)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid poll_id UUID")
|
|
rows = q.all()
|
|
return [{"id": str(r.id), "text": r.text, "poll_id": str(r.poll_id)} for r in rows]
|
|
|
|
|
|
@router.get("/{question_id}")
|
|
def get_question(question_id: str):
|
|
Session = get_session()
|
|
try:
|
|
qid = uuid.UUID(question_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid UUID")
|
|
with Session() as session:
|
|
q = session.get(Question, qid)
|
|
if not q:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
return {"id": str(q.id), "text": q.text, "poll_id": str(q.poll_id)}
|