[Feat] ORM 코드 수정

README.md

@@ -17,10 +17,8 @@ HexaCTF의 Container Control 프로젝트의 일부입니다.
 
 ## API
 
-|                          Endpoint                           | Method |     Description     |
-| :---------------------------------------------------------: | :----: | :-----------------: |
-|        [/v1/user-challenges](./md/user-challenge.md)        |  POST  |   Challenge 생성    |
-| [/v1/user-challenges/delete](./md/user-challenge-delete.md) |  POST  |   Challenge 삭제    |
-| [/v1/user-challenges/status](./md/user-challenge-status.md) |  GET   | Challenge 상태 조회 |
-
-
+|                           Endpoint                            | Method |     Description     |
+| :-----------------------------------------------------------: | :----: | :-----------------: |
+|        [/v1/user-challenges](./docs/user-challenge.md)        |  POST  |   Challenge 생성    |
+| [/v1/user-challenges/delete](./docs/user-challenge-delete.md) |  POST  |   Challenge 삭제    |
+| [/v1/user-challenges/status](./docs/user-challenge-status.md) |  GET   | Challenge 상태 조회 |

app.py

@@ -1,4 +1,4 @@
-from hexactf.factory import create_app
+from challenge_api.factory import create_app
 
 app = create_app()
 

challenge_api/api/challenge_api.py

@@ -0,0 +1,63 @@
+from json import JSONDecodeError
+from logging import log
+from flask import Blueprint, jsonify, request
+
+from challenge_api.exceptions.api_exceptions import InvalidRequest
+from challenge_api.exceptions.userchallenge_exceptions import UserChallengeCreationError, UserChallengeDeletionError, UserChallengeNotFoundError
+from challenge_api.db.repository import UserChallengesRepository, UserChallengeStatusRepository
+from challenge_api.extensions.k8s.client import K8sClient
+from challenge_api.utils.api_decorators import validate_request_body
+from challenge_api.objects.challenge_info import ChallengeInfo
+
+challenge_bp = Blueprint('challenge', __name__)
+
+@challenge_bp.route('', methods=['POST'])
+@validate_request_body('challenge_id', 'user_id')
+def create_challenge():
+    """사용자 챌린지 생성"""
+    # Challenge 관련 정보 가져오기 
+    res = request.get_json()
+    challenge_info = ChallengeInfo(**res)
+
+    # 챌린지 생성 
+    client = K8sClient()
+    endpoint = client.create(data=challenge_info)
+    if not endpoint:
+        raise UserChallengeCreationError(error_msg=f"Failed to create challenge {challenge_info.challenge_id} for user {challenge_info.name} : Endpoint did not exist")
+
+    return jsonify({'data' : {'port': endpoint}}), 200
+
+@challenge_bp.route('/delete', methods=['POST'])    
+@validate_request_body('challenge_id', 'user_id')
+def delete_userchallenges():
+    """사용자 챌린지 삭제"""
+    try:
+        res = request.get_json()
+        challenge_info = ChallengeInfo(**res)
+
+        # 사용자 챌린지 삭제 
+        client = K8sClient()
+        client.delete(challenge_info)
+
+        return jsonify({'message' : '챌린지가 정상적으로 삭제되었습니다.'}), 200
+    except JSONDecodeError as e:
+        log.error("Invalid request format")
+        raise InvalidRequest(error_msg=str(e)) from e
+
+@challenge_bp.route('/status', methods=['POST'])
+@validate_request_body('challenge_id', 'user_id')
+def get_userchallenge_status():
+    """사용자 챌린지 최근 상태 조회"""
+    try:
+        res = request.get_json()
+        challenge_info = ChallengeInfo(**res)
+
+        # 사용자 챌린지 상태 조회
+        userchallenge_repo = UserChallengesRepository()
+        userchallenge = userchallenge_repo.get_by_user_challenge_name(challenge_info.name)
+
+        repo = UserChallengeStatusRepository()
+        status = repo.get_recent_status(userchallenge.idx)
+        return jsonify({'data': {'port': status.port, 'status': status.status}}), 200
+    except Exception as e:
+        raise UserChallengeNotFoundError(error_msg=str(e))

hexactf/extensions/db/__init__.py → challenge_api/db/__init__.py

@@ -3,4 +3,4 @@
 __all__ = ['MariaDBConfig']
 
 from flask_sqlalchemy import SQLAlchemy
-from hexactf.extensions.db.config import MariaDBConfig
+from challenge_api.db.config import MariaDBConfig

hexactf/extensions/db/models.py → challenge_api/db/models.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timedelta
-from hexactf.extensions_manager import db
+from enum import Enum
+from challenge_api.extensions_manager import db
 from sqlalchemy import ForeignKey
 from sqlalchemy.orm import relationship
 
@@ -10,7 +11,8 @@ def current_time_kst():
 class Users(db.Model):
     __tablename__ = 'Users'
 
-    email = db.Column(db.String(255), primary_key=True, nullable=False, unique=True)
+    idx = db.Column(db.Integer, primary_key=True, nullable=False, autoincrement=True)
+    email = db.Column(db.String(255), nullable=False, unique=True)
     username = db.Column(db.String(20), nullable=False, unique=True)
     password = db.Column(db.String(255), nullable=False)
     homepage = db.Column(db.String(255), default='', nullable=False)
@@ -83,45 +85,37 @@ class Challenges(db.Model):
     unlockChallenges = db.Column(db.String(255), default='', nullable=False)
     isPersistence = db.Column(db.Boolean, default=False, nullable=False)
 
-
+# UserChallenges Table Model
 class UserChallenges(db.Model):
-    """사용자 챌린지 모델"""
     __tablename__ = 'UserChallenges'
 
     idx = db.Column(db.Integer, primary_key=True, autoincrement=True)
-    username = db.Column(db.String(20), ForeignKey('Users.username'), nullable=False)
-    C_idx = db.Column(db.Integer, ForeignKey('Challenges.idx'), nullable=False)
+    user_idx = db.Column(db.Integer, ForeignKey('Users.idx'), default=0, nullable=False)
+    C_idx = db.Column(db.Integer, db.ForeignKey('Challenges.idx'), nullable=False)
     userChallengeName = db.Column(db.String(255), nullable=False)
-    port = db.Column(db.Integer, nullable=False)
-    status = db.Column(db.String(50), default='None', nullable=False)
     createdAt = db.Column(db.DateTime, default=current_time_kst, nullable=False)
 
-    # 관계 설정
-    user = relationship('Users', backref='challenges')
-    challenge = relationship('Challenges', backref='user_challenges')
-
-    # __table_args__ = (
-    #     CheckConstraint('(port == 0) OR (port >= 15000 AND port <= 30000)', name='checkPort'),
-    # )
 
-    def __init__(self, username: str, C_idx: int, userChallengeName: str, port: int, status: str = 'None'):
-        self.username = username
-        self.C_idx = C_idx
-        self.userChallengeName = userChallengeName
-        self.port = port
-        self.status = status
+# UserChallenges_status Table Model
+class UserChallenges_status(db.Model):
+    __tablename__ = 'UserChallenges_status'
 
-    def to_dict(self) -> dict:
-        """모델을 딕셔너리로 변환"""
-        return {
-            'idx': self.idx,
-            'username': self.username,
-            'C_idx': self.C_idx,
-            'userChallengeName': self.userChallengeName,
-            'port': self.port,
-            'status': self.status,
-            'createdAt': self.createdAt.isoformat()
-        }
+    idx = db.Column(db.Integer, primary_key=True, autoincrement=True)
+    port = db.Column(db.Integer, nullable=False)
+    status = db.Column(db.String(20), nullable=False, default='None')
+    createdAt = db.Column(db.DateTime, default=current_time_kst, nullable=False)
+    userChallenge_idx = db.Column(db.Integer, db.ForeignKey('UserChallenges.idx'), nullable=False)
 
+# Submissions Table Model
+class Submissions(db.Model):
+    __tablename__ = 'Submissions'
 
-
+    idx = db.Column(db.Integer, primary_key=True, autoincrement=True)
+    C_idx = db.Column(db.Integer, ForeignKey('Challenges.idx'), nullable=False)
+    username = db.Column(db.String(20), ForeignKey('Users.username'), nullable=False)
+    teamName = db.Column(db.String(20), nullable=False)
+    title = db.Column(db.String(255), nullable=False)
+    type = db.Column(db.Boolean, nullable=False)
+    provided = db.Column(db.String(255), nullable=False)
+    currentStatus = db.Column(db.Boolean, default=True, nullable=False)
+    createdAt = db.Column(db.DateTime, default=current_time_kst, nullable=False)

challenge_api/db/repository/__init__.py

@@ -0,0 +1,9 @@
+from .challenge_repo import ChallengeRepository
+from .userchallenge_repo import UserChallengesRepository
+from .userchallenge_status_repo import UserChallengeStatusRepository
+
+__all__ = [
+    'ChallengeRepository',
+    'UserChallengesRepository',
+    'UserChallengeStatusRepository'
+]

challenge_api/db/repository/challenge_repo.py

@@ -0,0 +1,36 @@
+from challenge_api.db.models import Challenges
+from challenge_api.extensions_manager import db
+from challenge_api.objects.challenge_info import ChallengeInfo
+from challenge_api.exceptions.api_exceptions import InternalServerError
+from sqlalchemy.exc import SQLAlchemyError
+from typing import Optional
+
+class ChallengeRepository:
+
+    def __init__(self, session=None):
+        self.session = session or db.session
+
+    def is_exist(self, challenge_id: int) -> bool:
+        """
+        챌린지 아이디 존재 여부 확인
+
+        Args:
+            challenge_id (int): 챌린지 아이디
+
+        """
+        challenge = Challenges.query.get(challenge_id)
+        return challenge is not None
+
+    def get_challenge_name(challenge_id: int) -> Optional[str]:
+        """
+        챌린지 아이디로 챌린지 조회
+
+        Args:
+            challenge_id (int): 챌린지 아이디  
+
+        Returns:
+            str: 챌린지 이름
+        """
+        challenge = Challenges.query.get(challenge_id)
+        return challenge.title if challenge else None
+

challenge_api/db/repository/userchallenge_repo.py

@@ -0,0 +1,80 @@
+from challenge_api.db.models import UserChallenges
+from challenge_api.objects.challenge_info import ChallengeInfo
+from challenge_api.exceptions.api_exceptions import InternalServerError
+from sqlalchemy.exc import SQLAlchemyError
+from typing import Optional
+
+from challenge_api.extensions_manager import db
+
+class UserChallengesRepository:
+    def __init__(self, session=None):
+        self.session = session or db.session
+
+    def create(self, challenge_info: ChallengeInfo) -> Optional[UserChallenges]:
+        """새로운 사용자 챌린지 생성
+
+        Args:
+            challenge_name (ChallengeName): 챌린지 이름 객체
+
+        Returns:
+            Optional[UserChallenges]: 생성된 챌린지, 실패시 None
+
+        Raises:
+            InternalServerError: DB 작업 실패시
+        """
+        try:
+            challenge = UserChallenges(
+                C_idx=challenge_info.challenge_id,
+                user_idx=challenge_info.user_id,
+                userChallengeName=challenge_info.name,
+            )
+            self.session.add(challenge)
+            self.session.commit()
+            return challenge
+
+        except SQLAlchemyError as e:
+            self.session.rollback()
+            raise InternalServerError(error_msg=f"Error creating challenge in db: {e}") from e
+
+    def get_by_user_challenge_name(self, userChallengeName: str) -> Optional[UserChallenges]:
+        """
+        사용자 챌린지 이름 조회
+
+        Args:
+            userChallengeName (str): 사용자 챌린지 이름
+
+        Returns:
+            UserChallenges: 사용자 챌린지
+        """
+        user_challenge = UserChallenges.query.filter_by(userChallengeName=userChallengeName).first()
+        if not user_challenge:
+            return None
+        return user_challenge
+
+    def is_running(self, challenge: UserChallenges) -> bool:
+        """
+        챌린지 실행 여부 확인
+
+        Args:
+            challenge (UserChallenges): 사용자 챌린지
+
+        Returns:
+            bool: 챌린지 실행 여부
+        """
+        return challenge.status == 'Running'
+
+    def is_exist(self, challenge_info: ChallengeInfo) -> bool:
+        """챌린지 이름 존재 여부 확인
+
+        Args:
+            challenge_info (ChallengeInfo): 챌린지 이름 객체
+
+        Returns:
+            bool: 존재 여부
+        """
+        return (
+            UserChallenges.query
+            .filter_by(userChallengeName=challenge_info.name)
+            .first() is not None
+        )
+