♻ 重写 quiz 相关数据库操作

core/quiz/models.py

@@ -1,9 +1,25 @@
 from typing import List, Optional
 
+from sqlmodel import SQLModel, Field
+
 from models.baseobject import BaseObject
 from models.types import JSONDict
 
 
+class AnswerDB(SQLModel, table=True):
+    __tablename__ = 'answer'
+    id: Optional[int] = Field(default=None, primary_key=True)
+    question_id: Optional[int] = Field()
+    is_correct: Optional[bool] = Field()
+    text: Optional[str] = Field()
+
+
+class QuestionDB(SQLModel, table=True):
+    __tablename__ = 'question'
+    id: Optional[int] = Field(default=None, primary_key=True)
+    text: Optional[str] = Field()
+
+
 class Answer(BaseObject):
     def __init__(self, answer_id: int = 0, question_id: int = 0, is_correct: bool = True, text: str = ""):
         """Answer类

core/quiz/repositories.py

@@ -1,33 +1,56 @@
 from typing import List
 
+from sqlmodel import select
+
 from utils.mysql import MySQL
-from .models import Question, Answer
+from .models import QuestionDB, AnswerDB
 
 
 class QuizRepository:
     def __init__(self, mysql: MySQL):
         self.mysql = mysql
 
-    async def get_question_list(self) -> List[Question]:
-        pass
+    async def get_question_list(self) -> List[QuestionDB]:
+        async with self.mysql.Session() as session:
+            query = select(QuestionDB)
+            results = await session.exec(query)
+            questions = results.all()
+            return [question[0] for question in questions]
 
-    async def get_answer_form_question_id(self, question_id: int) -> List[Answer]:
-        pass
+    async def get_answer_form_question_id(self, question_id: int) -> List[AnswerDB]:
+        async with self.mysql.Session() as session:
+            query = select(AnswerDB).where(AnswerDB.question_id == question_id)
+            results = await session.exec(query)
+            answers = results.all()
+            return [answer[0] for answer in answers]
 
-    async def add_question(self, question: str):
-        pass
+    async def add_question(self, question: QuestionDB):
+        async with self.mysql.Session() as session:
+            session.add(question)
+            await session.commit()
 
-    async def get_question(self, question: str) -> Question:
-        pass
+    async def get_question_by_text(self, text: str) -> QuestionDB:
+        async with self.mysql.Session() as session:
+            query = select(QuestionDB).where(QuestionDB.text == text)
+            results = await session.exec(query)
+            question = results.first()
+            return question[0]
 
-    async def add_answer(self, question_id: int, is_correct: int, answer: str):
-        pass
+    async def add_answer(self, answer: AnswerDB):
+        async with self.mysql.Session() as session:
+            session.add(answer)
+            await session.commit()
 
-    async def delete_question(self, question_id: int):
-        pass
+    async def delete_question_by_id(self, question_id: int):
+        async with self.mysql.Session() as session:
+            statement = select(QuestionDB).where(QuestionDB.id == question_id)
+            results = await session.exec(statement)
+            question = results.one()
+            await session.delete(question)
 
-    async def delete_answer(self, answer_id: int):
-        pass
-
-    async def delete_admin(self, user_id: int):
-        pass
+    async def delete_answer_by_id(self, answer_id: int):
+        async with self.mysql.Session() as session:
+            statement = select(AnswerDB).where(AnswerDB.id == answer_id)
+            results = await session.exec(statement)
+            answer = results.one()
+            await session.delete(answer)