| from pydantic import BaseModel, Field
|
| from typing import Optional, AnyStr, List, Dict
|
| from ..utils.utils import get_current_time
|
| from ..configs.database_config import db
|
| from bson import ObjectId
|
|
|
| collection = db["word"]
|
|
|
|
|
| class Word(BaseModel):
|
| id: Optional[str] = Field(title="ID")
|
| word: str = Field(title="Word")
|
| answer: str = Field(title="Meaning")
|
| options: list = Field(title="Options")
|
| user_id: Optional[str] = Field(title="User ID")
|
|
|
| class Config:
|
| schema_extra = {
|
| "example": {
|
| "id": "666460100c23ec4225cb2bc3",
|
| "word": "Apple",
|
| "answer": "táo",
|
| "options": ["dừa", "thanh long", "ổi", "táo"],
|
| "user_id": "6661455703d07f73ba",
|
| }
|
| }
|
|
|
|
|
| class WordSchema:
|
| def __init__(
|
| self,
|
| id: AnyStr = None,
|
| word: AnyStr = "",
|
| answer: AnyStr = "",
|
| options: List = [AnyStr],
|
| user_id: AnyStr = "",
|
| created_at=get_current_time(),
|
| ):
|
| self.id = id
|
| self.word = word
|
| self.answer = answer
|
| self.options = options
|
| self.user_id = user_id
|
| self.created_at = created_at
|
|
|
| def to_dict(self):
|
| data_dict = {
|
| "word": self.word,
|
| "answer": self.answer,
|
| "options": self.options,
|
| "user_id": self.user_id,
|
| "created_at": self.created_at,
|
| }
|
| if self.id is not None:
|
| data_dict["_id"] = str(self.id)
|
| return data_dict
|
|
|
| @staticmethod
|
| def from_dict(data: Dict):
|
| return WordSchema(
|
| id=data.get("_id"),
|
| word=data.get("word"),
|
| answer=data.get("answer"),
|
| options=data.get("options"),
|
| user_id=data.get("user_id"),
|
| created_at=data.get("created_at"),
|
| )
|
|
|
| def create(self, user_id: str):
|
| word_dict = self.to_dict()
|
| word_dict["user_id"] = user_id
|
| collection.insert_one(word_dict)
|
|
|
| @staticmethod
|
| def read_all_words_by_user_id(user_id: str):
|
| data = collection.find({"user_id": user_id})
|
| return [WordSchema.from_dict(d).to_dict() for d in data]
|
|
|
| @staticmethod
|
| def read_word_by_id(word_id: str, user_id: str):
|
| data = collection.find_one({"_id": ObjectId(word_id), "user_id": user_id})
|
| return WordSchema.from_dict(data).to_dict()
|
|
|
| @staticmethod
|
| def check_existing_word(word: str, user_id: str):
|
| return collection.find_one({"word": word, "user_id": user_id})
|
|
|
| def update(self, word_id: str):
|
| collection.update_one(
|
| {"_id": ObjectId(word_id)},
|
| {"$set": self.to_dict()},
|
| )
|
|
|
| @staticmethod
|
| def delete(word_id: str):
|
| collection.delete_one({"_id": ObjectId(word_id)})
|
|
|
