feat: quiz create and quiz get done

app/blueprints/__init__.py

@@ -2,5 +2,14 @@ from .default import default_blueprint
 
 from .auth import auth_blueprint
 from .user import user_blueprint
+from .quiz import quiz_bp
+
+__all__ = [
+    "default_blueprint",
+    "auth_blueprint",
+    "user_blueprint",
+    "quiz_bp",
+]
+
 
 # from .user import user_blueprint

app/blueprints/quiz.py

@@ -0,0 +1,25 @@
+from flask import Blueprint, request
+from di_container import Container
+from dependency_injector.wiring import inject, Provide
+from controllers import QuizController
+
+
+quiz_bp = Blueprint(
+    "quiz",
+    __name__,
+)
+
+
+@quiz_bp.route("/quiz", methods=["POST"])
+@inject
+def create_quiz(controller: QuizController = Provide[Container.quiz_controller]):
+    reqBody = request.get_json()
+    return controller.create_quiz(reqBody)
+
+
+@quiz_bp.route("/quiz/<quiz_id>", methods=["GET"])
+@inject
+def get_quiz(
+    quiz_id: str, controller: QuizController = Provide[Container.quiz_controller]
+):
+    return controller.get_quiz(quiz_id)

app/controllers/__init__.py

@@ -1,2 +1,10 @@
 from .auth_controller import AuthController
 from .user_controller import UserController
+from .quiz_controller import QuizController
+
+
+__all__ = [
+    "AuthController",
+    "UserController",
+    "QuizController",
+]

app/controllers/quiz_controller.py

@@ -0,0 +1,52 @@
+from flask import jsonify
+from pydantic import ValidationError
+from schemas.requests import QuizCreateSchema
+from schemas.response import QuizCreationResponse
+from services import QuizService
+from helpers import make_response, make_error_response
+
+
+class QuizController:
+    def __init__(self, quiz_service: QuizService):
+        self.quiz_service = quiz_service
+
+    def get_quiz(self, quiz_id):
+        try:
+            result = self.quiz_service.get_quiz(quiz_id)
+            if not result:
+                return make_response(message="Quiz not found", status_code=404)
+            return make_response(message="Quiz Found", data=result.dict())
+        except Exception as e:
+            return make_error_response(e)
+
+    def create_quiz(self, quiz_data):
+        try:
+            quiz_obj = QuizCreateSchema(**quiz_data)
+            quiz_id = self.quiz_service.create_quiz(quiz_obj)
+            return make_response(
+                message="Quiz created",
+                data=QuizCreationResponse(quiz_id=quiz_id),
+                status_code=201,
+            )
+        except ValidationError as e:
+            return make_response(e.errors(), status_code=400)
+        except Exception as e:
+            return make_error_response(e)
+
+    # def update_quiz(self, quiz_id, quiz_data):
+    #     try:
+    #         quiz_obj = QuizUpdateSchema(**quiz_data)
+    #         success = self.quiz_service.update_quiz(
+    #             quiz_id, quiz_obj.dict(exclude_unset=True)
+    #         )
+    #         if not success:
+    #             return jsonify({"error": "Quiz not found or update failed"}), 400
+    #         return jsonify({"message": "Quiz updated"}), 200
+    #     except ValidationError as e:
+    #         return jsonify({"error": "Validation error", "detail": e.errors()}), 400
+
+    def delete_quiz(self, quiz_id):
+        success = self.quiz_service.delete_quiz(quiz_id)
+        if not success:
+            return jsonify({"error": "Quiz not found"}), 400
+        return jsonify({"message": "Quiz deleted"}), 200

app/controllers/user_controller.py

@@ -5,6 +5,7 @@ from schemas import RegisterSchema
 from pydantic import ValidationError
 from schemas import ResponseSchema
 from exception import AlreadyExistException
+from helpers import make_response
 
 
 class UserController:
@@ -16,26 +17,17 @@ class UserController:
             request_data = request.get_json()
             register_data = RegisterSchema(**request_data)
             self.user_service.register_user(register_data)
-            return jsonify(ResponseSchema(message="Register Success").model_dump()), 200
+            return make_response("Register Success")
 
         except ValidationError as e:
             current_app.logger.error(f"Validation error: {e}")
             response = ResponseSchema(message="Invalid input", data=None, meta=None)
-            return jsonify(response.model_dump()), 400
+            return make_response("Invalid input", status_code=400)
 
         except AlreadyExistException as e:
-            return (
+            return make_response("User already exists", status_code=409)
-                jsonify(
-                    ResponseSchema(message=str(e), data=None, meta=None).model_dump()
-                ),
-                409,
-            )
-
         except Exception as e:
             current_app.logger.error(
                 f"Error during Google login: {str(e)}", exc_info=True
             )
-            response = ResponseSchema(
+            return make_response("Internal server error", status_code=500)
-                message="Internal server error", data=None, meta=None
-            )
-            return jsonify(response.model_dump()), 500

app/di_container.py

@@ -4,6 +4,9 @@ from repositories.user_repository import UserRepository
 from services import UserService, AuthService
 from controllers import AuthController
 from flask_pymongo import PyMongo
+from repositories import QuizRepository
+from services import QuizService
+from controllers import QuizController
 
 
 class Container(containers.DeclarativeContainer):
@@ -13,11 +16,14 @@ class Container(containers.DeclarativeContainer):
 
     # repository
     user_repository = providers.Factory(UserRepository, mongo.provided.db)
+    quiz_repository = providers.Factory(QuizRepository, mongo.provided.db)
 
     # services
     auth_service = providers.Factory(AuthService, user_repository)
     user_service = providers.Factory(UserService, user_repository)
+    quiz_service = providers.Factory(QuizService, quiz_repository)
 
     # controllers
     auth_controller = providers.Factory(AuthController, user_service, auth_service)
     user_controller = providers.Factory(UserController, user_service)
+    quiz_controller = providers.Factory(QuizController, quiz_service)

app/exception/__init__.py

@@ -1,2 +1,10 @@
 from .auth_exception import AuthException
 from .already_exist_exception import AlreadyExistException
+from .data_not_found_exception import DataNotFoundException
+
+
+__all__ = [
+    "AuthException",
+    "AlreadyExistException",
+    "DataNotFoundException",
+]

app/exception/data_not_found_exception.py

@@ -0,0 +1,8 @@
+from .base_exception import BaseExceptionTemplate
+
+
+class DataNotFoundException(BaseExceptionTemplate):
+    """Exception for data not found"""
+
+    def __init__(self, message: str = "Data Not Found"):
+        super().__init__(message, status_code=404)

app/helpers/__init__.py

@@ -0,0 +1,4 @@
+from .response_helper import make_response, make_error_response
+
+
+__all__ = ["make_response", "make_error_response"]

app/helpers/response_helper.py

@@ -0,0 +1,27 @@
+from flask import jsonify, current_app
+from typing import Optional, Union
+from schemas import ResponseSchema
+
+
+def make_response(
+    message: str,
+    data: Optional[dict] = None,
+    meta: Optional[dict] = None,
+    status_code: int = 200,
+):
+    response = ResponseSchema(message=message, data=data, meta=meta)
+    return jsonify(response.model_dump()), status_code
+
+
+def make_error_response(
+    err: Union[Exception, str],
+    log_message: Optional[str] = None,
+    status_code: int = 500,
+):
+    """Logs the error and returns a standardized error response"""
+    error_message = str(err) if isinstance(err, Exception) else err
+    log_msg = log_message or f"An error occurred: {error_message}"
+    current_app.logger.error(log_msg, exc_info=True)
+
+    response = ResponseSchema(message="Internal server error", data=None, meta=None)
+    return jsonify(response.model_dump()), status_code

app/main.py

@@ -1,8 +1,7 @@
-from blueprints import default_blueprint
 from di_container import Container
 from configs import Config, LoggerConfig
 from flask import Flask
-from blueprints import auth_blueprint, user_blueprint
+from blueprints import auth_blueprint, user_blueprint, quiz_bp, default_blueprint
 from database import init_db
 import logging
 
@@ -26,11 +25,13 @@ def createApp() -> Flask:
 
     container.wire(modules=["blueprints.auth"])
     container.wire(modules=["blueprints.user"])
+    container.wire(modules=["blueprints.quiz"])
 
     # Register Blueprints
     app.register_blueprint(default_blueprint)
     app.register_blueprint(auth_blueprint, url_prefix="/api")
     app.register_blueprint(user_blueprint, url_prefix="/api")
+    app.register_blueprint(quiz_bp, url_prefix="/api")
 
     return app
 

app/mapper/__init__.py

@@ -1 +1,8 @@
 from .user_mapper import UserMapper
+from .quiz_mapper import map_quiz_entity_to_schema
+
+
+__all__ = [
+    "UserMapper",
+    "map_quiz_entity_to_schema",
+]

app/mapper/quiz_mapper.py

@@ -0,0 +1,26 @@
+from models import QuizEntity, QuestionItemEntity
+from schemas import QuizGetSchema, QuestionItemSchema
+
+
+def map_question_entity_to_schema(entity: QuestionItemEntity) -> QuestionItemSchema:
+    return QuestionItemSchema(
+        question=entity.question,
+        target_answer=entity.target_answer,
+        duration=entity.duration,
+        type=entity.type,
+    )
+
+
+def map_quiz_entity_to_schema(entity: QuizEntity) -> QuizGetSchema:
+    return QuizGetSchema(
+        author_id=entity.author_id,
+        title=entity.title,
+        description=entity.description,
+        is_public=entity.is_public,
+        date=entity.date.strftime("%Y-%m-%d %H:%M:%S") if entity.date else None,
+        total_quiz=entity.total_quiz or 0,
+        limit_duration=entity.limit_duration or 0,
+        question_listings=[
+            map_question_entity_to_schema(q) for q in entity.question_listings or []
+        ],
+    )

app/models/__init__.py

@@ -1,5 +1,12 @@
 # app/models/__init__.py
-from .entities import UserEntity
+from .entities import UserEntity, QuizEntity, QuestionItemEntity
 from .login import UserResponseModel
 
-__all__ = ["UserEntity", "UserDTO", "UserResponseModel"]
+
+__all__ = [
+    "UserEntity",
+    "UserDTO",
+    "UserResponseModel",
+    "QuizEntity",
+    "QuestionItemEntity",
+]

app/models/entities/__init__.py

@@ -1,7 +1,11 @@
 from .user_entity import UserEntity
 from .base import PyObjectId
+from .quiz_entity import QuizEntity
+from .question_item_entity import QuestionItemEntity
 
 __all__ = [
     "UserEntity",
     "PyObjectId",
+    "QuizEntity",
+    "QuestionItemEntity",
 ]

app/models/entities/question_item_entity.py

@@ -0,0 +1,12 @@
+from typing import Optional, List
+from pydantic import BaseModel
+from datetime import datetime
+from .base import PyObjectId
+
+
+class QuestionItemEntity(BaseModel):
+    _id: Optional[PyObjectId] = None
+    question: str
+    target_answer: str
+    duration: int
+    type: str  # "isian" | "true_false"

app/models/entities/quiz_entity.py

@@ -0,0 +1,21 @@
+from typing import Optional
+from pydantic import BaseModel
+from datetime import datetime
+from .base import PyObjectId
+from .question_item_entity import QuestionItemEntity
+
+
+class QuizEntity(BaseModel):
+    _id: Optional[PyObjectId] = None
+    author_id: Optional[str] = None
+    title: str
+    description: Optional[str] = None
+    is_public: bool = False
+    date: Optional[datetime] = None
+    total_quiz: Optional[int] = 0
+    limit_duration: Optional[int] = 0
+    question_listings: Optional[list[QuestionItemEntity]] = []
+
+    class Config:
+        arbitrary_types_allowed = True
+        json_encoders = {PyObjectId: str}

app/repositories/__init__.py

@@ -1 +1,8 @@
 from .user_repository import UserRepository
+from .quiz_repositroy import QuizRepository
+
+
+__all__ = [
+    "UserRepository",
+    "QuizRepository",
+]

app/repositories/quiz_repositroy.py

@@ -0,0 +1,33 @@
+from bson import ObjectId
+from typing import List, Optional
+from models import QuizEntity
+
+
+class QuizRepository:
+    def __init__(self, db):
+        self.collection = db.quiz
+
+    def create(self, quiz: QuizEntity) -> str:
+        quiz_dict = quiz.dict(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 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 delete(self, quiz_id: str) -> bool:
+        result = self.collection.delete_one({"_id": ObjectId(quiz_id)})
+        return result.deleted_count > 0

app/schemas/__init__.py

@@ -1,3 +1,15 @@
 from .login_schema import LoginSchema
 from .basic_response_schema import ResponseSchema, MetaSchema
 from .requests import RegisterSchema
+from .response import QuizCreationResponse, QuizGetSchema, QuestionItemSchema
+
+
+__all__ = [
+    "LoginSchema",
+    "ResponseSchema",
+    "MetaSchema",
+    "RegisterSchema",
+    "QuizCreationResponse",
+    "QuizGetSchema",
+    "QuestionItemSchema",
+]

app/schemas/requests/__init__.py

@@ -1 +1,13 @@
 from .register_schema import RegisterSchema
+
+from .quiz import (
+    QuestionItemSchema,
+    QuizCreateSchema,
+)
+
+
+__all__ = [
+    "RegisterSchema",
+    "QuestionItemSchema",
+    "QuizCreateSchema",
+]

app/schemas/requests/quiz/__init__.py

@@ -0,0 +1,7 @@
+from .quiz_item_schema import QuestionItemSchema
+from .create_quiz_schema import QuizCreateSchema
+
+__all__ = [
+    "QuestionItemSchema",
+    "QuizCreateSchema",
+]

app/schemas/requests/quiz/create_quiz_schema.py

@@ -0,0 +1,15 @@
+from typing import List, Optional
+from pydantic import BaseModel, Field
+from datetime import datetime
+from .quiz_item_schema import QuestionItemSchema
+
+
+class QuizCreateSchema(BaseModel):
+    title: str
+    description: Optional[str] = None
+    is_public: bool = False
+    date: Optional[datetime] = None
+    total_quiz: Optional[int] = 0
+    limit_duration: Optional[int] = 0
+    author_id: Optional[str] = None
+    question_listings: Optional[List[QuestionItemSchema]] = []

app/schemas/requests/quiz/quiz_item_schema.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class QuestionItemSchema(BaseModel):
+    question: str
+    target_answer: str
+    duration: int
+    type: str

app/schemas/response/__init__.py

@@ -0,0 +1,9 @@
+from .quiz.quiz_creation_response import QuizCreationResponse
+from .quiz.quiz_get_response import QuizGetSchema
+from .quiz.question_item_schema import QuestionItemSchema
+
+__all__ = [
+    "QuizCreationResponse",
+    "QuizGetSchema",
+    "QuestionItemSchema",
+]

app/schemas/response/quiz/question_item_schema.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class QuestionItemSchema(BaseModel):
+    question: str
+    target_answer: str
+    duration: int
+    type: str

app/schemas/response/quiz/quiz_creation_response.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class QuizCreationResponse(BaseModel):
+    quiz_id: str

app/schemas/response/quiz/quiz_get_response.py

@@ -0,0 +1,15 @@
+from typing import List, Optional
+from pydantic import BaseModel
+from datetime import datetime
+from .question_item_schema import QuestionItemSchema
+
+
+class QuizGetSchema(BaseModel):
+    author_id: str
+    title: str
+    description: Optional[str] = None
+    is_public: bool = False
+    date: Optional[str] = None
+    total_quiz: int = 0
+    limit_duration: int = 0
+    question_listings: List[QuestionItemSchema] = []

app/services/__init__.py

@@ -1,2 +1,10 @@
 from .auth_service import AuthService
 from .user_service import UserService
+from .quiz_service import QuizService
+
+
+__all__ = [
+    "AuthService",
+    "UserService",
+    "QuizService",
+]

app/services/lstm_service.py

@@ -1,58 +1,61 @@
-from keras.models import load_model
+# from keras.models import load_model
-import pickle
+# from tensorflow.keras.preprocessing.sequence import pad_sequences
+# from nummpy import nps
+
+# import pickle
 
 
-class LSTMService:
+# class LSTMService:
 
-    def predict(self, input_data, maxlen=50):
+#     def predict(self, input_data, maxlen=50):
 
-        with open("QC/tokenizers.pkl", "rb") as f:
+#         with open("QC/tokenizers.pkl", "rb") as f:
-            tokenizers = pickle.load(f)
+#             tokenizers = pickle.load(f)
 
-        model = load_model("QC/lstm_qg.keras")
+#         model = load_model("QC/lstm_qg.keras")
 
-        tok_token = tokenizers["token"]
+#         tok_token = tokenizers["token"]
-        tok_ner = tokenizers["ner"]
+#         tok_ner = tokenizers["ner"]
-        tok_srl = tokenizers["srl"]
+#         tok_srl = tokenizers["srl"]
-        tok_q = tokenizers["question"]
+#         tok_q = tokenizers["question"]
-        tok_a = tokenizers["answer"]
+#         tok_a = tokenizers["answer"]
-        tok_type = tokenizers["type"]
+#         tok_type = tokenizers["type"]
 
-        # Prepare input
+#         # Prepare input
-        tokens = input_data["tokens"]
+#         tokens = input_data["tokens"]
-        ner = input_data["ner"]
+#         ner = input_data["ner"]
-        srl = input_data["srl"]
+#         srl = input_data["srl"]
 
-        x_tok = pad_sequences(
+#         x_tok = pad_sequences(
-            [tok_token.texts_to_sequences([tokens])[0]], maxlen=maxlen, padding="post"
+#             [tok_token.texts_to_sequences([tokens])[0]], maxlen=maxlen, padding="post"
-        )
+#         )
-        x_ner = pad_sequences(
+#         x_ner = pad_sequences(
-            [tok_ner.texts_to_sequences([ner])[0]], maxlen=maxlen, padding="post"
+#             [tok_ner.texts_to_sequences([ner])[0]], maxlen=maxlen, padding="post"
-        )
+#         )
-        x_srl = pad_sequences(
+#         x_srl = pad_sequences(
-            [tok_srl.texts_to_sequences([srl])[0]], maxlen=maxlen, padding="post"
+#             [tok_srl.texts_to_sequences([srl])[0]], maxlen=maxlen, padding="post"
-        )
+#         )
 
-        # Predict
+#         # Predict
-        pred_q, pred_a, pred_type = model.predict([x_tok, x_ner, x_srl])
+#         pred_q, pred_a, pred_type = model.predict([x_tok, x_ner, x_srl])
-        pred_q_ids = np.argmax(pred_q[0], axis=-1)
+#         pred_q_ids = np.argmax(pred_q[0], axis=-1)
-        pred_a_ids = np.argmax(pred_a[0], axis=-1)
+#         pred_a_ids = np.argmax(pred_a[0], axis=-1)
-        pred_type_id = np.argmax(pred_type[0])
+#         pred_type_id = np.argmax(pred_type[0])
 
-        # Decode
+#         # Decode
-        index2word_q = {v: k for k, v in tok_q.word_index.items()}
+#         index2word_q = {v: k for k, v in tok_q.word_index.items()}
-        index2word_a = {v: k for k, v in tok_a.word_index.items()}
+#         index2word_a = {v: k for k, v in tok_a.word_index.items()}
-        index2word_q[0] = "<PAD>"
+#         index2word_q[0] = "<PAD>"
-        index2word_a[0] = "<PAD>"
+#         index2word_a[0] = "<PAD>"
 
-        decoded_q = [index2word_q[i] for i in pred_q_ids if i != 0]
+#         decoded_q = [index2word_q[i] for i in pred_q_ids if i != 0]
-        decoded_a = [index2word_a[i] for i in pred_a_ids if i != 0]
+#         decoded_a = [index2word_a[i] for i in pred_a_ids if i != 0]
 
-        index2type = {v - 1: k for k, v in tok_type.word_index.items()}
+#         index2type = {v - 1: k for k, v in tok_type.word_index.items()}
-        decoded_type = index2type.get(pred_type_id, "unknown")
+#         decoded_type = index2type.get(pred_type_id, "unknown")
 
-        return {
+#         return {
-            "question": " ".join(decoded_q),
+#             "question": " ".join(decoded_q),
-            "answer": " ".join(decoded_a),
+#             "answer": " ".join(decoded_a),
-            "type": decoded_type,
+#             "type": decoded_type,
-        }
+#         }

app/services/quiz_service.py

@@ -0,0 +1,25 @@
+from repositories import QuizRepository
+from schemas import QuizGetSchema
+from exception import DataNotFoundException
+from mapper import map_quiz_entity_to_schema
+
+
+class QuizService:
+    def __init__(self, quiz_repository=QuizRepository):
+        self.quiz_repository = quiz_repository
+
+    def get_quiz(self, quiz_id) -> QuizGetSchema:
+        data = self.quiz_repository.get_by_id(quiz_id)
+        if data is None:
+            raise DataNotFoundException("Quiz not found")
+
+        return map_quiz_entity_to_schema(data)
+
+    def create_quiz(self, quiz_data):
+        return self.quiz_repository.create(quiz_data)
+
+    def update_quiz(self, quiz_id, quiz_data):
+        return self.quiz_repository.update(quiz_id, quiz_data)
+
+    def delete_quiz(self, quiz_id):
+        return self.quiz_repository.delete(quiz_id)