94e78a9e77
- bd/__init__.py: добавлены get_engine() и get_session() — один SQLAlchemy engine на весь процесс - make_engine(): pool_size=5, max_overflow=5, pool_timeout=30, pool_recycle=1800 - Все 17 route-файлов: убраны локальные get_session()/make_engine(), импорт из bd - transfer_crud.py: _get_session() → get_session() из bd - init_data_base.py: get_engine() вместо make_engine() - Результат: 78 idle соединений → 1
91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import List, Optional
|
|
import uuid
|
|
|
|
from bd import get_session
|
|
|
|
from bd.tables.poll import Poll
|
|
from bd.tables.question import Question
|
|
from bd.tables.choice import Choice
|
|
from route.auth_utils import require_admin_key
|
|
|
|
router = APIRouter(tags=["polls"], prefix="/polls")
|
|
|
|
|
|
class ChoiceIn(BaseModel):
|
|
text: str
|
|
position: Optional[int]
|
|
|
|
|
|
class QuestionIn(BaseModel):
|
|
text: str
|
|
type: Optional[str] = "single"
|
|
position: Optional[int]
|
|
choices: Optional[List[ChoiceIn]]
|
|
|
|
|
|
class PollIn(BaseModel):
|
|
title: str
|
|
description: Optional[str]
|
|
questions: Optional[List[QuestionIn]]
|
|
|
|
|
|
|
|
|
|
@router.post("/", status_code=201, dependencies=[Depends(require_admin_key)])
|
|
def create_poll(payload: PollIn):
|
|
Session = get_session()
|
|
with Session() as session:
|
|
poll = Poll(title=payload.title, description=payload.description)
|
|
# polls are anonymous; no author_id stored
|
|
session.add(poll)
|
|
# questions and choices
|
|
if payload.questions:
|
|
for q in payload.questions:
|
|
question = Question(text=q.text, type=q.type or "single", position=q.position, poll=poll)
|
|
session.add(question)
|
|
if q.choices:
|
|
for idx, c in enumerate(q.choices):
|
|
choice = Choice(text=c.text, position=c.position if c.position is not None else idx, question=question)
|
|
session.add(choice)
|
|
session.commit()
|
|
session.refresh(poll)
|
|
return {"id": str(poll.id)}
|
|
|
|
|
|
@router.get("/{poll_id}")
|
|
def get_poll(poll_id: str):
|
|
Session = get_session()
|
|
try:
|
|
pid = uuid.UUID(poll_id)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid UUID")
|
|
with Session() as session:
|
|
poll = session.get(Poll, pid)
|
|
if not poll:
|
|
raise HTTPException(status_code=404, detail="Poll not found")
|
|
data = {
|
|
"id": str(poll.id),
|
|
"title": poll.title,
|
|
"description": poll.description,
|
|
"questions": [
|
|
{
|
|
"id": str(q.id),
|
|
"text": q.text,
|
|
"type": q.type,
|
|
"choices": [{"id": str(c.id), "text": c.text} for c in q.choices]
|
|
}
|
|
for q in poll.questions
|
|
],
|
|
}
|
|
return data
|
|
|
|
|
|
@router.get("/")
|
|
def list_polls():
|
|
Session = get_session()
|
|
with Session() as session:
|
|
rows = session.query(Poll).all()
|
|
return [{"id": str(p.id), "title": p.title, "description": p.description} for p in rows]
|