diff --git a/src/blueprints/__init__.py b/src/blueprints/__init__.py index a966bed..c979444 100644 --- a/src/blueprints/__init__.py +++ b/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"] \ No newline at end of file diff --git a/src/blueprints/bp_category.py b/src/blueprints/bp_category.py new file mode 100644 index 0000000..072780f --- /dev/null +++ b/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("/") +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 + ) \ No newline at end of file diff --git a/src/constants.py b/src/constants.py index 979787e..2e57e80 100644 --- a/src/constants.py +++ b/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" diff --git a/src/models/__init__.py b/src/models/__init__.py index 8ccadba..f35f28f 100644 --- a/src/models/__init__.py +++ b/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, ] \ No newline at end of file diff --git a/src/models/category_model/category.py b/src/models/category_model/category.py new file mode 100644 index 0000000..77be2de --- /dev/null +++ b/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' 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 \ No newline at end of file diff --git a/src/repositories/category/category_repository.py b/src/repositories/category/category_repository.py new file mode 100644 index 0000000..ae5d011 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/src/services/category/category_service.py b/src/services/category/category_service.py new file mode 100644 index 0000000..6182d16 --- /dev/null +++ b/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) diff --git a/src/utils/swagger_config.py b/src/utils/swagger_config.py index 3e93adf..9624ebd 100644 --- a/src/utils/swagger_config.py +++ b/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__ = [