115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import uuid
|
|
|
|
from bd import Settings
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from bd.tables.choice import Choice
|
|
from bd.tables.question import Question
|
|
|
|
router = APIRouter(tags=["choices"], prefix="/choices")
|
|
|
|
|
|
class ChoiceCreate(BaseModel):
|
|
question_id: str
|
|
text: str
|
|
position: Optional[int]
|
|
|
|
|
|
class ChoiceUpdate(BaseModel):
|
|
text: Optional[str]
|
|
position: Optional[int]
|
|
|
|
|
|
def get_session():
|
|
settings = Settings()
|
|
engine = create_engine(settings.DATABASE_URL_syncpg, future=True)
|
|
return sessionmaker(bind=engine, future=True)
|
|
|
|
|
|
@router.post("/", status_code=201)
|
|
def create_choice(payload: ChoiceCreate):
|
|
Session = get_session()
|
|
try:
|
|
qid = uuid.UUID(payload.question_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid question UUID")
|
|
with Session() as session:
|
|
question = session.get(Question, qid)
|
|
if not question:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
ch = Choice(question_id=qid, text=payload.text, position=payload.position)
|
|
session.add(ch)
|
|
session.commit()
|
|
session.refresh(ch)
|
|
return {"id": str(ch.id)}
|
|
|
|
|
|
@router.put("/{choice_id}")
|
|
def update_choice(choice_id: str, payload: ChoiceUpdate):
|
|
Session = get_session()
|
|
try:
|
|
cid = uuid.UUID(choice_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid UUID")
|
|
with Session() as session:
|
|
ch = session.get(Choice, cid)
|
|
if not ch:
|
|
raise HTTPException(status_code=404, detail="Choice not found")
|
|
if payload.text is not None:
|
|
ch.text = payload.text
|
|
if payload.position is not None:
|
|
ch.position = payload.position
|
|
session.add(ch)
|
|
session.commit()
|
|
session.refresh(ch)
|
|
return {"id": str(ch.id)}
|
|
|
|
|
|
@router.delete("/{choice_id}", status_code=204)
|
|
def delete_choice(choice_id: str):
|
|
Session = get_session()
|
|
try:
|
|
cid = uuid.UUID(choice_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid UUID")
|
|
with Session() as session:
|
|
ch = session.get(Choice, cid)
|
|
if not ch:
|
|
raise HTTPException(status_code=404, detail="Choice not found")
|
|
session.delete(ch)
|
|
session.commit()
|
|
return {}
|
|
|
|
|
|
@router.get("/")
|
|
def list_choices(question_id: Optional[str] = None):
|
|
Session = get_session()
|
|
with Session() as session:
|
|
q = session.query(Choice)
|
|
if question_id:
|
|
try:
|
|
qid = uuid.UUID(question_id)
|
|
q = q.filter(Choice.question_id == qid)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid question_id UUID")
|
|
rows = q.all()
|
|
return [{"id": str(r.id), "text": r.text, "question_id": str(r.question_id)} for r in rows]
|
|
|
|
|
|
@router.get("/{choice_id}")
|
|
def get_choice(choice_id: str):
|
|
Session = get_session()
|
|
try:
|
|
cid = uuid.UUID(choice_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid UUID")
|
|
with Session() as session:
|
|
ch = session.get(Choice, cid)
|
|
if not ch:
|
|
raise HTTPException(status_code=404, detail="Choice not found")
|
|
return {"id": str(ch.id), "text": ch.text, "question_id": str(ch.question_id)}
|