[FEAT] (카레고리 로직): 카테고리 서비스 구현 완료

v0.1.4 (2025-11-27)
- 카테고리 관련 API 구현 완료.

src/blueprints/__init__.py

@@ -12,6 +12,6 @@ def register_blueprints(app: Flask) -> None:
     app.register_blueprint(api_blueprint, url_prefix="/api/v1")
     
     # Flask-RESTX namespace에 리소스 등록
-    from . import bp_auth, bp_users
+    from . import bp_auth, bp_users, bp_category
 
 __all__ = ["register_blueprints"]

src/blueprints/bp_category.py

@@ -0,0 +1,142 @@
+"""
+카테고리 관련 API 엔드포인트
+"""
+
+from flask import request, g
+from flask_restx import Resource
+from flask_jwt_extended import jwt_required
+
+from utils.response import *
+from utils.swagger_config import category_ns
+from utils.decorators import load_current_user
+
+from models.category_model import category_swagger
+
+from services.category.category_service import CategoryService
+
+@category_ns.route("/")
+class Categories(Resource):
+    @category_ns.doc(
+        id="get_categories",
+        summary="카테고리 목록 조회",
+        description="카테고리 목록을 조회합니다.",
+        tags=["Category"],
+        security=[{"Bearer": []}]
+    )
+    @category_ns.response(200, "Success - 카테고리 목록 조회 완료")
+    @category_ns.response(401, "인증에 실패하였습니다.")
+    @category_ns.response(500, "서버 내부 오류")
+    @jwt_required()
+    def get(self):
+        """카테고리 목록 조회"""
+        limit = request.args.get("limit", type=int)
+        offset = request.args.get("offset", type=int, default=0)
+        
+        categories = CategoryService.get_all_categories(limit=limit, offset=offset)
+        return success_response(
+            data=[category.to_dict() for category in categories],
+            message="카테고리 목록 조회 성공"
+        )
+    
+    @category_ns.doc(
+        id="create_category",
+        summary="카테고리 생성",
+        description="카테고리를 생성합니다.",
+        tags=["Category"],
+        security=[{"Bearer": []}]
+    )
+    @category_ns.expect(category_swagger.create_category)
+    @category_ns.response(201, "Success - 카테고리 생성 완료")
+    @category_ns.response(401, "인증에 실패하였습니다.")
+    @category_ns.response(409, "이미 존재하는 카테고리명")
+    @category_ns.response(500, "서버 내부 오류")
+    @jwt_required()
+    @load_current_user
+    def post(self):
+        """카테고리 생성"""
+        request_data = request.get_json()
+
+        # 토큰에서 가져온 user_uuid를 request_data에 추가
+        user_uuid = g.current_user.user_uuid
+
+        category = CategoryService.create_category(request_data, user_uuid)
+        return success_response(
+            data=category.to_dict(),
+            message="카테고리 생성이 완료되었습니다",
+            status_code=201
+        )
+
+
+@category_ns.route("/<string:category_uuid>")
+class CategoryDetail(Resource):
+    @category_ns.doc(
+        id="get_category",
+        summary="카테고리 조회",
+        description="UUID로 특정 카테고리를 조회합니다.",
+        tags=["Category"],
+        security=[{"Bearer": []}],
+        params={"category_uuid": {"description": "카테고리 UUID", "type": "string"}}
+    )
+    @category_ns.response(201, "Success - 카테고리 조회 성공")
+    @category_ns.response(401, "인증에 실패하였습니다.")
+    @category_ns.response(409, "조회에 실패하였습니다.")
+    @category_ns.response(500, "서버 내부 오류")
+    @jwt_required()
+    def get(self, category_uuid):
+        """카테고리 조회"""
+        category = CategoryService.get_category_by_uuid(category_uuid)
+
+        if not category:
+            return not_found_response("Category")
+
+        return success_response(
+            data=category.to_dict(),
+            message="카테고리 조회 성공"
+        )
+    
+    @category_ns.doc(
+        id="update_category",
+        summary="카테고리 업데이트",
+        description="UUID로 특정 카테고리를 업데이트합니다.",
+        tags=["Category"],
+        security=[{"Bearer": []}],
+        params={"category_uuid": {"description": "카테고리 UUID", "type": "string"}}
+    )
+    @category_ns.expect(category_swagger.update_category)
+    @category_ns.response(200, "Success - 카테고리 업데이트 성공")
+    @category_ns.response(401, "인증에 실패하였습니다.")
+    @category_ns.response(409, "카테고리 업데이트에 실패하였습니다.")
+    @category_ns.response(500, "서버 내부 오류")
+    @jwt_required()
+    def put(self, category_uuid):
+        """카테고리 업데이트"""
+        request_data = request.get_json()
+
+        category = CategoryService.update_category(category_uuid, request_data)
+        return success_response(
+            data=category.to_dict(),
+            message="카테고리 업데이트가 완료되었습니다",
+            status_code=200
+        )
+
+    @category_ns.doc(
+        id="delete_category",
+        summary="카테고리 삭제",
+        description="UUID로 특정 카테고리를 삭제합니다.",
+        tags=["Category"],
+        security=[{"Bearer": []}],
+        params={"category_uuid": {"description": "카테고리 UUID", "type": "string"}}
+    )
+    @category_ns.response(200, "Success - 카테고리 삭제 성공")
+    @category_ns.response(401, "인증에 실패하였습니다.")
+    @category_ns.response(409, "카테고리 삭제에 실패하였습니다.")
+    @category_ns.response(500, "서버 내부 오류")
+    @jwt_required()
+    def delete(self, category_uuid):
+        """카테고리 삭제"""
+        CategoryService.delete_category(category_uuid)
+
+        return success_response(
+            message="카테고리 삭제가 완료되었습니다",
+            status_code=200
+        )

src/constants.py

@@ -1,6 +1,6 @@
 # API INFO
-API_VERSION = "v0.1.3"
-API_DATE = "2025-11-16"
+API_VERSION = "v0.1.4"
+API_DATE = "2025-11-17"
 
 # BASE DATE FORMAT
 DATE_FORMAT = "%Y-%m-%d %H:%M:%S %z"

src/models/__init__.py

@@ -5,10 +5,12 @@ Models 레이어
 # User
 from .user_model import users, user_ips, user_dto, auth_swagger
 
+# Category
+from .category_model import category, category_dto, category_swagger
+
 __all__ = [
     # User
-    users,
-    user_ips,
-    user_dto,
-    auth_swagger
+    users, user_ips, user_dto, auth_swagger,
+    # Category
+    category, category_dto, category_swagger,
 ]

src/models/category_model/category.py

@@ -0,0 +1,63 @@
+"""
+카테고리 모델
+"""
+
+import uuid
+from typing import Dict, Any
+from extensions import db
+from utils.func import get_timestamp_ms
+import constants
+
+class Category(db.Model):
+    """카테고리 모델"""
+    
+    __tablename__ = "category"
+    
+    # 기본 필드
+    category_uuid = db.Column(db.String(255), primary_key=True, nullable=False, comment="카타게로 UUID")
+    category_name = db.Column(db.String(255), unique=True, nullable=False, index=True, comment="카테고리명")
+    description = db.Column(db.String(500), nullable=True, comment="카테고리 설명")
+    order_no = db.Column(db.Integer, nullable=True, comment="표시 순서")
+    is_used = db.Column(db.SmallInteger, nullable=False, default=constants.IS_Y, comment="사용 여부 (0:미사용, 1:사용)")
+    
+    # 편집자
+    created_by = db.Column(db.String(255), nullable=True, comment="생성자 UUID")
+    updated_by = db.Column(db.String(255), nullable=True, comment="수정자 UUID")
+    deleted_by = db.Column(db.String(255), nullable=True, comment="삭제자 UUID")
+    
+    # 타임스탬프 (BIGINT - Unix Timestamp in milliseconds)
+    created_at = db.Column(db.BigInteger, nullable=True, comment="생성일")
+    updated_at = db.Column(db.BigInteger, nullable=True, comment="수정일")
+    deleted_at = db.Column(db.BigInteger, nullable=True, comment="삭제일")
+
+    def __init__(self, category_name: str, description: str = None, created_by: str = None):
+        """카테고리 초기화"""
+        self.category_uuid = str(uuid.uuid4())
+        self.category_name = category_name
+        self.description = description
+        self.is_used = constants.IS_Y
+        self.created_by = created_by
+        self.created_at = int(get_timestamp_ms())
+    
+    def update_timestamp(self) -> None:
+        """타임스탬프 업데이트"""
+        self.updated_at = int(get_timestamp_ms())
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """딕셔너리로 변환"""
+        return {
+            "category_uuid": self.category_uuid,
+            "category_name": self.category_name,
+            "description": self.description,
+            "order_no": self.order_no,
+            "is_used": self.is_used,
+            "created_by": self.created_by,
+            "updated_by": self.updated_by,
+            "deleted_by": self.deleted_by,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+            "deleted_at": self.deleted_at,
+        }
+    
+    def __repr__(self) -> str:
+        return f'<Category uuid={self.category_uuid}, category_name={self.category_name}, is_used={self.is_used}'

src/models/category_model/category_dto.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, field_validator
+from typing import Optional
+
+class CreateCategoryDTO(BaseModel):
+    """카테고리 생성 DTO"""
+
+    # 필수
+    category_name: str
+
+    description: Optional[str] = None
+
+    class Config:
+        strict = True
+
+
+class UpdateCategoryDTO(BaseModel):
+    """카테고리 업데이트 DTO"""
+
+    category_name: Optional[str] = None
+    description: Optional[str] = None
+
+    class Config:
+        strict = True

src/models/category_model/category_swagger.py

@@ -0,0 +1,16 @@
+"""
+카테고리 Swagger 모델 정의
+"""
+
+from utils.swagger_config import category_ns
+from flask_restx import fields
+
+create_category = category_ns.model("CreateCategory", {
+    "category_name": fields.String(required=True, description="카테고리명", example="AI"),
+    "description": fields.String(required=False, description="카테고리 설명", example="AI 관련 주제입니다.")
+})
+
+update_category = category_ns.model("UpdateCategory", {
+    "category_name": fields.String(required=False, description="카테고리명", example="AI"),
+    "description": fields.String(required=False, description="카테고리 설명", example="AI 관련 주제입니다.")
+})

src/repositories/__init__.py

@@ -9,9 +9,14 @@ from .base_repository import BaseRepository
 from .user.user_repository import UserRepository
 from .user.user_ips_repository import UserIpsRepository
 
+# Category
+from .category.category_repository import CategoryRepository
+
 __all__ = [
     # Base
     BaseRepository,
     # User
-    UserRepository, UserIpsRepository
+    UserRepository, UserIpsRepository,
+    # Category
+    CategoryRepository,
 ]

src/repositories/base_repository.py

@@ -69,4 +69,20 @@ class BaseRepository(ABC, Generic[T]):
             return self.session.query(self.model).count()
         except SQLAlchemyError as e:
             logger.error(f"{self.model.__name__} 개수 조회 실패: {str(e)}")
+            raise
+    
+    def find_all(self, limit: Optional[int] = None, offset: int = 0) -> List[T]:
+        """Find all entities with optional pagination."""
+        try:
+            query = self.session.query(self.model)
+
+            if offset > 0:
+                query = query.offset(offset)
+
+            if limit:
+                query = query.limit(limit)
+
+            return query.all()
+        except SQLAlchemyError as e:
+            logger.error(f"Error finding all {self.model.__name__}: {e}")
             raise

src/repositories/category/category_repository.py

@@ -0,0 +1,36 @@
+"""
+카테고리 Repository
+"""
+
+from typing import Optional, Type
+
+from repositories.base_repository import BaseRepository
+from models.category_model.category import Category
+
+class CategoryRepository(BaseRepository[Category]):
+    """카테고리 Repository"""
+    
+    @property
+    def model(self) -> Type[Category]:
+        """모델 반환"""
+        return Category
+    
+    def find_by_uuid(self, category_uuid: str) -> Optional[Category]:
+        """UUID로 조회"""
+        return self.session.query(Category).filter_by(category_uuid=category_uuid).one_or_none()
+
+    def find_by_name(self, category_name: str) -> Optional[Category]:
+        """카테고리명으로 조회"""
+        return self.session.query(Category).filter_by(category_name=category_name).one_or_none()
+
+    def exists_by_name(self, category_name: str) -> bool:
+        """카테고리명으로 카테고리 존재 여부 확인"""
+        return self.session.query(
+            self.session.query(Category).filter_by(category_name=category_name).exists()
+        ).scalar()
+
+    def exists_by_uuid(self, category_uuid: str) -> bool:
+        """UUID로 카테고리 존재 여부 확인"""
+        return self.session.query(
+            self.session.query(Category).filter_by(category_uuid=category_uuid).exists()
+        ).scalar()

src/services/category/category_service.py

@@ -0,0 +1,122 @@
+"""
+카테고리 서비스
+"""
+
+from typing import Dict, Any, Optional, List
+
+from models.category_model.category import Category
+from models.category_model.category_dto import CreateCategoryDTO
+
+from repositories import CategoryRepository
+
+from utils.db_decorators import transactional
+from extensions import custom_logger
+import settings
+import constants
+
+logger = custom_logger(f"{settings.LOG_PREFIX}_category_service")
+
+class CategoryService:
+    """
+    카테고리 서비스
+    """
+    
+    @transactional
+    def create_category(category_data: Dict[str, Any], user_uuid: str) -> Optional[Category]:
+        """카테고리 생성"""
+        
+        create_category_dto = CreateCategoryDTO(**category_data)
+        
+        category_repo = CategoryRepository()
+        
+        # 중복 검사
+        if category_repo.exists_by_name(create_category_dto.category_name):
+            return None
+        
+        # 카테고리 생성
+        category = Category(
+            category_name=create_category_dto.category_name,
+            description=create_category_dto.description,
+            created_by=user_uuid
+        )
+        
+        category = category_repo.create(category)
+        logger.info(f"카테고리 생성 완료: {category.category_uuid}")
+        
+        return category
+    
+    def get_category_by_uuid(category_uuid: str) -> Optional[Category]:
+        """UUID로 카테고리 조회"""
+        category_repo = CategoryRepository()
+        return category_repo.find_by_uuid(category_uuid)
+    
+    def get_category_by_name(category_name: str) -> Optional[Category]:
+        """카테고리명으로 카테고리 조회"""
+        category_repo = CategoryRepository()
+        return category_repo.find_by_name(category_name)
+    
+    def get_all_categories(limit: Optional[int] = None, offset: int = 0) -> List[Category]:
+        """모든 카테고리 조회"""
+        category_repo = CategoryRepository()
+        return category_repo.find_all(limit=limit, offset=offset)
+    
+    @transactional
+    def update_category(category_uuid: str, update_data: Dict[str, Any]) -> Category:
+        """카테고리 업데이트"""
+        category_repo = CategoryRepository()
+        
+        # 카테고리 조회
+        category = category_repo.find_by_uuid(category_uuid)
+        
+        if not category:
+            return None
+        
+        # 이름 변경 시 중복 검사
+        if 'category_name' in update_data and update_data['category_name'] != category.category_name:
+            if category_repo.exists_by_name(update_data['category_name']):
+                return None
+            category.category_name = update_data['category_name']
+
+        # 설명 업데이트
+        if 'description' in update_data:
+            category.description = update_data['description']
+
+        # 타임스탬프 업데이트
+        if hasattr(category, 'update_timestamp'):
+            category.update_timestamp()
+
+        category = category_repo.update(category)
+        logger.info(f"Category updated: {category.category_name} ({category_uuid})")
+
+        return category
+    
+    @transactional
+    def delete_category(category_uuid: str) -> bool:
+        """카테고리 삭제"""
+        category_repo = CategoryRepository()
+
+        # 카테고리 조회
+        category = category_repo.find_by_uuid(category_uuid)
+
+        if not category:
+            return None
+
+        category.is_used = constants.IS_N
+        
+        # 삭제
+        success = category_repo.update(category)
+
+        if success:
+            logger.info(f"Category deleted: {category.category_name} ({category_uuid})")
+
+        return success
+
+    def count_categories() -> int:
+        """카테고리 총 개수 조회"""
+        category_repo = CategoryRepository()
+        return category_repo.count()
+
+    def exists_by_name(category_name: str) -> bool:
+        """카테고리명 존재 여부 확인"""
+        category_repo = CategoryRepository()
+        return category_repo.exists_by_name(category_name)

src/utils/swagger_config.py

@@ -32,6 +32,7 @@ api = Api(
 # 네임 스페이스 정의 및 API 추가
 auth_ns = api.namespace("auth", description="인증 API", path="/auth")
 user_ns = api.namespace("users", description="유저 API", path="/users")
+category_ns = api.namespace("categories", description="카테고리 API", path="/categories")
 
 # 네임 스페이스 정의 및 API에 추가
 __all__ = [