You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
77 lines
2.4 KiB
77 lines
2.4 KiB
from fastapi import APIRouter, Depends, HTTPException |
|
from sqlalchemy import select |
|
from sqlalchemy.ext.asyncio import AsyncSession |
|
from sqlalchemy.orm import selectinload |
|
|
|
from app.database import get_db |
|
from app.models.test import Answer, Question, Test |
|
from app.schemas.test import TestCreate, TestListItem, TestOut |
|
|
|
router = APIRouter(prefix="/api/tests", tags=["tests"]) |
|
|
|
|
|
@router.get("", response_model=list[TestListItem]) |
|
async def list_tests(db: AsyncSession = Depends(get_db)): |
|
result = await db.execute( |
|
select(Test) |
|
.options(selectinload(Test.questions)) |
|
.where(Test.is_active == True) |
|
.order_by(Test.created_at.desc()) |
|
) |
|
tests = result.scalars().all() |
|
|
|
items = [] |
|
for test in tests: |
|
item = TestListItem.model_validate(test) |
|
item.questions_count = len(test.questions) |
|
items.append(item) |
|
|
|
return items |
|
|
|
|
|
@router.get("/{test_id}", response_model=TestOut) |
|
async def get_test(test_id: int, db: AsyncSession = Depends(get_db)): |
|
result = await db.execute( |
|
select(Test) |
|
.options(selectinload(Test.questions).selectinload(Question.answers)) |
|
.where(Test.id == test_id, Test.is_active == True) |
|
) |
|
test = result.scalar_one_or_none() |
|
if not test: |
|
raise HTTPException(status_code=404, detail="Тест не найден") |
|
return test |
|
|
|
|
|
@router.post("", response_model=TestOut, status_code=201) |
|
async def create_test(data: TestCreate, db: AsyncSession = Depends(get_db)): |
|
test = Test( |
|
title=data.title, |
|
description=data.description, |
|
passing_score=data.passing_score, |
|
time_limit=data.time_limit, |
|
allow_navigation_back=data.allow_navigation_back, |
|
) |
|
db.add(test) |
|
await db.flush() |
|
|
|
for order, q_data in enumerate(data.questions): |
|
question = Question(test_id=test.id, text=q_data.text, order=order) |
|
db.add(question) |
|
await db.flush() |
|
|
|
for a_data in q_data.answers: |
|
db.add(Answer( |
|
question_id=question.id, |
|
text=a_data.text, |
|
is_correct=a_data.is_correct, |
|
)) |
|
|
|
await db.commit() |
|
|
|
# Перезагружаем с вложенными связями |
|
result = await db.execute( |
|
select(Test) |
|
.options(selectinload(Test.questions).selectinload(Question.answers)) |
|
.where(Test.id == test.id) |
|
) |
|
return result.scalar_one()
|
|
|