72 lines
1.9 KiB
Python
72 lines
1.9 KiB
Python
from sqlalchemy import select, delete
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
from models.library import Mistake, Subject
|
|
from schemas.mistake import MistakeResponse, MistakeRequest
|
|
from service import subject_service
|
|
|
|
|
|
async def list_mistakes(subject: str, db: AsyncSession) -> list[MistakeResponse]:
|
|
subject_id = await subject_service.get_subject_by_name(db, subject)
|
|
stmt = (
|
|
select(Mistake)
|
|
.options(joinedload(Mistake.subject))
|
|
.where(Mistake.subject_id == subject_id)
|
|
.order_by(Mistake.id.desc())
|
|
)
|
|
|
|
mistakes = (await db.execute(stmt)).scalars().unique().all()
|
|
|
|
return [
|
|
MistakeResponse(
|
|
id=mistake.id,
|
|
type=mistake.type,
|
|
question=mistake.question,
|
|
options=mistake.options,
|
|
answer=mistake.answer,
|
|
)
|
|
for mistake in mistakes
|
|
]
|
|
|
|
|
|
async def add_mistakes(db: AsyncSession, subject: str, data: list[MistakeRequest]):
|
|
subject_id = await subject_service.get_subject_by_name(db, subject)
|
|
|
|
subject = (
|
|
await db.execute(
|
|
select(Subject).where(Subject.id == subject_id)
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
if not subject:
|
|
raise ValueError("Subject not found")
|
|
|
|
mistakes = [
|
|
Mistake(
|
|
subject_id=subject_id,
|
|
type=item.type,
|
|
question=item.question,
|
|
options=item.options,
|
|
answer=item.answer,
|
|
)
|
|
for item in data
|
|
]
|
|
|
|
try:
|
|
db.add_all(mistakes)
|
|
await db.flush()
|
|
await db.commit()
|
|
return True
|
|
except SQLAlchemyError:
|
|
await db.rollback()
|
|
raise
|
|
|
|
|
|
async def remove_mistake(db: AsyncSession, mistake_id: int) -> bool:
|
|
await db.execute(delete(Mistake).where(Mistake.id == mistake_id))
|
|
await db.commit()
|
|
|
|
return True
|