feat: adding history endpoint
This commit is contained in:
parent
0773609fda
commit
58dcf0b15a
|
@ -3,12 +3,14 @@ from .default import default_blueprint
|
|||
from .auth import auth_blueprint
|
||||
from .user import user_blueprint
|
||||
from .quiz import quiz_bp
|
||||
from .history import history_blueprint
|
||||
|
||||
__all__ = [
|
||||
"default_blueprint",
|
||||
"auth_blueprint",
|
||||
"user_blueprint",
|
||||
"quiz_bp",
|
||||
"history_blueprint",
|
||||
]
|
||||
|
||||
|
||||
|
|
|
@ -0,0 +1,14 @@
|
|||
from flask import Blueprint
|
||||
from controllers import HistoryController
|
||||
from di_container import Container
|
||||
from dependency_injector.wiring import inject, Provide
|
||||
|
||||
history_blueprint = Blueprint("history", __name__)
|
||||
|
||||
|
||||
@history_blueprint.route("/<user_id>", methods=["GET"])
|
||||
@inject
|
||||
def user_history(
|
||||
user_id: str, controller: HistoryController = Provide[Container.history_controller]
|
||||
):
|
||||
return controller.get_quiz_by_user(user_id)
|
|
@ -1,10 +1,12 @@
|
|||
from .auth_controller import AuthController
|
||||
from .user_controller import UserController
|
||||
from .quiz_controller import QuizController
|
||||
from .history_controller import HistoryController
|
||||
|
||||
|
||||
__all__ = [
|
||||
"AuthController",
|
||||
"UserController",
|
||||
"QuizController",
|
||||
"HistoryController",
|
||||
]
|
||||
|
|
|
@ -0,0 +1,15 @@
|
|||
from services import HistoryService
|
||||
from helpers import make_error_response, make_response
|
||||
|
||||
|
||||
class HistoryController:
|
||||
|
||||
def __init__(self, history_service: HistoryService):
|
||||
self.history_service = history_service
|
||||
|
||||
def get_quiz_by_user(self, user_id: str):
|
||||
try:
|
||||
data = self.history_service.get_history_by_user_id(user_id)
|
||||
return make_response(message="retrive history data", data=data)
|
||||
except Exception as e:
|
||||
return make_error_response(e)
|
|
@ -1,7 +1,18 @@
|
|||
from dependency_injector import containers, providers
|
||||
from controllers import UserController, AuthController, QuizController
|
||||
from controllers import (
|
||||
UserController,
|
||||
AuthController,
|
||||
QuizController,
|
||||
HistoryController,
|
||||
)
|
||||
from repositories import UserRepository, QuizRepository, UserAnswerRepository
|
||||
from services import UserService, AuthService, QuizService, AnswerService
|
||||
from services import (
|
||||
UserService,
|
||||
AuthService,
|
||||
QuizService,
|
||||
AnswerService,
|
||||
HistoryService,
|
||||
)
|
||||
|
||||
|
||||
class Container(containers.DeclarativeContainer):
|
||||
|
@ -25,7 +36,14 @@ class Container(containers.DeclarativeContainer):
|
|||
user_repository,
|
||||
)
|
||||
|
||||
history_service = providers.Factory(
|
||||
HistoryService,
|
||||
quiz_repository,
|
||||
answer_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, answer_service)
|
||||
history_controller = providers.Factory(HistoryController, history_service)
|
||||
|
|
17
app/main.py
17
app/main.py
|
@ -1,7 +1,18 @@
|
|||
import sys
|
||||
import os
|
||||
|
||||
sys.path.append(os.path.dirname(__file__))
|
||||
|
||||
from di_container import Container
|
||||
from configs import Config, LoggerConfig
|
||||
from flask import Flask
|
||||
from blueprints import auth_blueprint, user_blueprint, quiz_bp, default_blueprint
|
||||
from blueprints import (
|
||||
auth_blueprint,
|
||||
user_blueprint,
|
||||
quiz_bp,
|
||||
default_blueprint,
|
||||
history_blueprint,
|
||||
)
|
||||
from database import init_db
|
||||
import logging
|
||||
|
||||
|
@ -32,6 +43,7 @@ def createApp() -> Flask:
|
|||
"blueprints.auth",
|
||||
"blueprints.user",
|
||||
"blueprints.quiz",
|
||||
"blueprints.history",
|
||||
]
|
||||
)
|
||||
|
||||
|
@ -40,6 +52,7 @@ def createApp() -> Flask:
|
|||
app.register_blueprint(auth_blueprint, url_prefix="/api")
|
||||
app.register_blueprint(user_blueprint, url_prefix="/api")
|
||||
app.register_blueprint(quiz_bp, url_prefix="/api/quiz")
|
||||
app.register_blueprint(history_blueprint, url_prefix="/api/history")
|
||||
|
||||
for rule in app.url_map.iter_rules():
|
||||
print(f"Route: {rule} -> Methods: {rule.methods}")
|
||||
|
@ -47,6 +60,8 @@ def createApp() -> Flask:
|
|||
return app
|
||||
|
||||
|
||||
# app = createApp()
|
||||
|
||||
if __name__ == "__main__":
|
||||
app = createApp()
|
||||
app.run(host="0.0.0.0", debug=Config.DEBUG)
|
||||
|
|
|
@ -1,12 +1,12 @@
|
|||
from typing import Optional
|
||||
from pydantic import BaseModel
|
||||
from pydantic import BaseModel, Field
|
||||
from datetime import datetime
|
||||
from .base import PyObjectId
|
||||
from .question_item_entity import QuestionItemEntity
|
||||
|
||||
|
||||
class QuizEntity(BaseModel):
|
||||
_id: Optional[PyObjectId] = None
|
||||
id: Optional[PyObjectId] = Field(alias="_id")
|
||||
author_id: Optional[str] = None
|
||||
title: str
|
||||
description: Optional[str] = None
|
||||
|
@ -18,4 +18,5 @@ class QuizEntity(BaseModel):
|
|||
|
||||
class Config:
|
||||
arbitrary_types_allowed = True
|
||||
populate_by_name = True
|
||||
json_encoders = {PyObjectId: str}
|
||||
|
|
|
@ -8,7 +8,7 @@ from .base import PyObjectId
|
|||
|
||||
|
||||
class UserAnswerEntity(BaseModel):
|
||||
_id: Optional[PyObjectId] = None
|
||||
id: Optional[PyObjectId] = Field(alias="_id")
|
||||
session_id: Optional[str]
|
||||
quiz_id: str
|
||||
user_id: str
|
||||
|
@ -16,7 +16,6 @@ class UserAnswerEntity(BaseModel):
|
|||
answers: List[AnswerItemEntity]
|
||||
total_score: int
|
||||
total_correct: int
|
||||
total_questions: int
|
||||
|
||||
class Config:
|
||||
populate_by_name = True
|
||||
|
|
|
@ -23,6 +23,10 @@ class UserAnswerRepository:
|
|||
)
|
||||
return list(result)
|
||||
|
||||
def get_by_user(self, user_id: str) -> list[UserAnswerEntity]:
|
||||
result = self.collection.find({"user_id": user_id})
|
||||
return [UserAnswerEntity(**doc) for doc in result]
|
||||
|
||||
def get_by_session(self, session_id: str) -> List[dict]:
|
||||
result = self.collection.find({"session_id": ObjectId(session_id)})
|
||||
return list(result)
|
||||
|
|
|
@ -2,12 +2,13 @@ from bson import ObjectId
|
|||
from typing import List, Optional
|
||||
from models import QuizEntity
|
||||
from pymongo.database import Database
|
||||
from pymongo.collection import Collection
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class QuizRepository:
|
||||
def __init__(self, db: Database):
|
||||
self.collection = db.quiz
|
||||
self.collection: Collection = db.quiz
|
||||
|
||||
def create(self, quiz: QuizEntity) -> str:
|
||||
quiz_dict = quiz.dict(by_alias=True, exclude_none=True)
|
||||
|
@ -20,6 +21,17 @@ class QuizRepository:
|
|||
return QuizEntity(**data)
|
||||
return None
|
||||
|
||||
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)
|
||||
print(datas)
|
||||
|
||||
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]:
|
||||
|
|
|
@ -2,10 +2,12 @@ from .quiz.quiz_creation_response import QuizCreationResponse
|
|||
from .quiz.quiz_get_response import QuizGetSchema
|
||||
from .quiz.question_item_schema import QuestionItemSchema
|
||||
from .quiz.quiz_data_rsp_schema import UserQuizListResponse
|
||||
from .history.history_response import HistoryResultSchema
|
||||
|
||||
__all__ = [
|
||||
"QuizCreationResponse",
|
||||
"QuizGetSchema",
|
||||
"QuestionItemSchema",
|
||||
"UserQuizListResponse"
|
||||
"UserQuizListResponse",
|
||||
"HistoryResultSchema",
|
||||
]
|
||||
|
|
|
@ -0,0 +1,12 @@
|
|||
from pydantic import BaseModel, Field
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class HistoryResultSchema(BaseModel):
|
||||
quiz_id: str = Field(..., description="ID dari kuis")
|
||||
answer_id: str = Field(..., description="ID dari jawaban")
|
||||
title: str = Field(..., description="Judul kuis")
|
||||
description: Optional[str] = Field(None, description="Deskripsi kuis")
|
||||
total_correct: int = Field(..., description="Jumlah jawaban benar")
|
||||
total_question: int = Field(..., description="Total soal dalam kuis")
|
||||
date: str
|
|
@ -2,11 +2,12 @@ from .auth_service import AuthService
|
|||
from .user_service import UserService
|
||||
from .quiz_service import QuizService
|
||||
from .answer_service import AnswerService
|
||||
|
||||
from .history_service import HistoryService
|
||||
|
||||
__all__ = [
|
||||
"AuthService",
|
||||
"UserService",
|
||||
"QuizService",
|
||||
"AnswerService",
|
||||
"HistoryService",
|
||||
]
|
||||
|
|
|
@ -0,0 +1,39 @@
|
|||
from repositories import UserAnswerRepository, QuizRepository
|
||||
from schemas.response import HistoryResultSchema
|
||||
|
||||
|
||||
class HistoryService:
|
||||
def __init__(
|
||||
self,
|
||||
quiz_repository: QuizRepository,
|
||||
answer_repository: UserAnswerRepository,
|
||||
):
|
||||
self.quiz_repository = quiz_repository
|
||||
self.answer_repository = answer_repository
|
||||
|
||||
def get_history_by_user_id(self, user_id: str):
|
||||
answer_data = self.answer_repository.get_by_user(user_id)
|
||||
if not answer_data:
|
||||
return []
|
||||
|
||||
quiz_ids = [asn.quiz_id for asn in answer_data]
|
||||
quiz_data = self.quiz_repository.get_by_ids(quiz_ids)
|
||||
quiz_map = {str(quiz.id): quiz for quiz in quiz_data}
|
||||
|
||||
result = []
|
||||
for answer in answer_data:
|
||||
quiz = quiz_map.get(answer.quiz_id)
|
||||
if quiz:
|
||||
result.append(
|
||||
HistoryResultSchema(
|
||||
quiz_id=str(quiz.id),
|
||||
answer_id=str(answer.id),
|
||||
title=quiz.title,
|
||||
description=quiz.description,
|
||||
total_correct=answer.total_correct,
|
||||
total_question=quiz.total_quiz,
|
||||
date=answer.answered_at.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
)
|
||||
)
|
||||
|
||||
return result
|
Loading…
Reference in New Issue