129 lines
4.3 KiB
Python
129 lines
4.3 KiB
Python
from bson import ObjectId
|
|
from typing import List, Optional
|
|
from app.models.entities import QuizEntity
|
|
from pymongo.database import Database
|
|
from pymongo.collection import Collection
|
|
|
|
|
|
class QuizRepository:
|
|
def __init__(self, db: Database):
|
|
self.collection: Collection = db.quiz
|
|
|
|
def create(self, quiz: QuizEntity) -> str:
|
|
quiz_dict = quiz.model_dump(by_alias=True, exclude_none=True)
|
|
result = self.collection.insert_one(quiz_dict)
|
|
return str(result.inserted_id)
|
|
|
|
def get_by_id(self, quiz_id: str) -> Optional[QuizEntity]:
|
|
data = self.collection.find_one({"_id": ObjectId(quiz_id)})
|
|
if data:
|
|
return QuizEntity(**data)
|
|
return None
|
|
|
|
def search_by_title_or_category(
|
|
self, keyword: str, subject_id: Optional[str], page: int, page_size: int
|
|
) -> List[QuizEntity]:
|
|
skip = (page - 1) * page_size
|
|
|
|
query_conditions = [
|
|
{"is_public": True},
|
|
{
|
|
"$or": [
|
|
{"title": {"$regex": keyword, "$options": "i"}},
|
|
# {"category": {"$regex": keyword, "$options": "i"}},
|
|
]
|
|
},
|
|
]
|
|
|
|
if subject_id:
|
|
query_conditions.append({"subject_id": subject_id})
|
|
|
|
cursor = (
|
|
self.collection.find({"$and": query_conditions}).skip(skip).limit(page_size)
|
|
)
|
|
|
|
return [QuizEntity(**doc) for doc in cursor]
|
|
|
|
def count_by_search(self, keyword: str) -> int:
|
|
return self.collection.count_documents(
|
|
{
|
|
"$or": [
|
|
{"title": {"$regex": keyword, "$options": "i"}},
|
|
{"category": {"$regex": keyword, "$options": "i"}},
|
|
]
|
|
}
|
|
)
|
|
|
|
def get_by_ids(self, quiz_ids: List[str]) -> Optional[List[QuizEntity]]:
|
|
object_ids = [ObjectId(qid) for qid in quiz_ids]
|
|
cursor = self.collection.find({"_id": {"$in": object_ids}})
|
|
datas = list(cursor)
|
|
|
|
if not datas:
|
|
return None
|
|
|
|
return [QuizEntity(**data) for data in datas]
|
|
|
|
def get_by_user_id(
|
|
self, user_id: str, page: int = 1, page_size: int = 10
|
|
) -> List[QuizEntity]:
|
|
skip = (page - 1) * page_size
|
|
cursor = (
|
|
self.collection.find({"author_id": user_id}).skip(skip).limit(page_size)
|
|
)
|
|
|
|
return [QuizEntity(**data) for data in cursor]
|
|
|
|
def get_all(self, skip: int = 0, limit: int = 10) -> List[QuizEntity]:
|
|
cursor = self.collection.find().skip(skip).limit(limit)
|
|
return [QuizEntity(**doc) for doc in cursor]
|
|
|
|
def update(self, quiz_id: str, update_data: dict) -> bool:
|
|
result = self.collection.update_one(
|
|
{"_id": ObjectId(quiz_id)}, {"$set": update_data}
|
|
)
|
|
return result.modified_count > 0
|
|
|
|
def update_user_playing(self, quiz_id: str, total_user: int) -> bool:
|
|
result = self.collection.update_one(
|
|
{"_id": ObjectId(quiz_id)}, {"$set": {"total_user_playing": total_user}}
|
|
)
|
|
return result.modified_count > 0
|
|
|
|
def delete(self, quiz_id: str) -> bool:
|
|
result = self.collection.delete_one({"_id": ObjectId(quiz_id)})
|
|
return result.deleted_count > 0
|
|
|
|
def count_by_user_id(self, user_id: str) -> int:
|
|
return self.collection.count_documents({"author_id": user_id})
|
|
|
|
def get_top_played_quizzes(
|
|
self, page: int = 1, limit: int = 3, is_public: bool = True
|
|
) -> List[QuizEntity]:
|
|
skip = (page - 1) * limit
|
|
cursor = (
|
|
self.collection.find({"is_public": is_public})
|
|
.sort("total_user_playing", -1)
|
|
.skip(skip)
|
|
.limit(limit)
|
|
)
|
|
return [QuizEntity(**doc) for doc in cursor]
|
|
|
|
def get_random_quizzes(self, limit: int = 3) -> List[QuizEntity]:
|
|
pipeline = [
|
|
{"$match": {"is_public": True}},
|
|
{"$sample": {"size": limit}},
|
|
]
|
|
cursor = self.collection.aggregate(pipeline)
|
|
return [QuizEntity(**doc) for doc in cursor]
|
|
|
|
def get_random_quizzes_by_subjects(
|
|
self, subject_ids: List[str], limit: int = 3
|
|
) -> List[QuizEntity]:
|
|
pipeline = [
|
|
{"$match": {"subject_id": {"$in": subject_ids}, "is_public": True}},
|
|
{"$sample": {"size": limit}},
|
|
]
|
|
cursor = self.collection.aggregate(pipeline)
|
|
return [QuizEntity(**doc) for doc in cursor]
|