mirror of
https://github.com/PaiGramTeam/PamGram.git
synced 2024-11-16 12:02:16 +00:00
233e7ab58d
Co-authored-by: luoshuijs <luoshuijs@outlook.com> Co-authored-by: Karako <karakohear@gmail.com> Co-authored-by: xtaodada <xtao@xtaolink.cn>
70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
from typing import List
|
|
|
|
from core.base_service import BaseService
|
|
from core.dependence.redisdb import RedisDB
|
|
from core.services.quiz.models import Answer, Question
|
|
|
|
__all__ = ("QuizCache",)
|
|
|
|
|
|
class QuizCache(BaseService.Component):
|
|
def __init__(self, redis: RedisDB):
|
|
self.client = redis.client
|
|
self.question_qname = "quiz:question"
|
|
self.answer_qname = "quiz:answer"
|
|
|
|
async def get_all_question(self) -> List[Question]:
|
|
temp_list = []
|
|
qname = self.question_qname + "id_list"
|
|
data_list = [self.question_qname + f":{question_id}" for question_id in await self.client.lrange(qname, 0, -1)]
|
|
data = await self.client.mget(data_list)
|
|
for i in data:
|
|
temp_list.append(Question.parse_raw(i))
|
|
return temp_list
|
|
|
|
async def get_all_question_id_list(self) -> List[str]:
|
|
qname = self.question_qname + ":id_list"
|
|
return await self.client.lrange(qname, 0, -1)
|
|
|
|
async def get_one_question(self, question_id: int) -> Question:
|
|
qname = f"{self.question_qname}:{question_id}"
|
|
data = await self.client.get(qname)
|
|
json_data = str(data, encoding="utf-8")
|
|
return Question.parse_raw(json_data)
|
|
|
|
async def get_one_answer(self, answer_id: int) -> Answer:
|
|
qname = f"{self.answer_qname}:{answer_id}"
|
|
data = await self.client.get(qname)
|
|
json_data = str(data, encoding="utf-8")
|
|
return Answer.parse_raw(json_data)
|
|
|
|
async def add_question(self, question_list: List[Question] = None) -> int:
|
|
if not question_list:
|
|
return 0
|
|
for question in question_list:
|
|
await self.client.set(f"{self.question_qname}:{question.question_id}", question.json())
|
|
question_id_list = [question.question_id for question in question_list]
|
|
await self.client.lpush(f"{self.question_qname}:id_list", *question_id_list)
|
|
return await self.client.llen(f"{self.question_qname}:id_list")
|
|
|
|
async def del_all_question(self):
|
|
keys = await self.client.keys(f"{self.question_qname}*")
|
|
if keys is not None:
|
|
for key in keys:
|
|
await self.client.delete(key)
|
|
|
|
async def del_all_answer(self):
|
|
keys = await self.client.keys(f"{self.answer_qname}*")
|
|
if keys is not None:
|
|
for key in keys:
|
|
await self.client.delete(key)
|
|
|
|
async def add_answer(self, answer_list: List[Answer] = None) -> int:
|
|
if not answer_list:
|
|
return 0
|
|
for answer in answer_list:
|
|
await self.client.set(f"{self.answer_qname}:{answer.answer_id}", answer.json())
|
|
answer_id_list = [answer.answer_id for answer in answer_list]
|
|
await self.client.lpush(f"{self.answer_qname}:id_list", *answer_id_list)
|
|
return await self.client.llen(f"{self.answer_qname}:id_list")
|