48 KiB

Raw Blame History

디자인 패턴 기반 리팩토링 제안서

1. 현재 아키텍처 분석

1.1 현재 구조

app/
├── {module}/
│   ├── models.py           # SQLAlchemy 모델
│   ├── schemas/            # Pydantic 스키마
│   ├── services/           # 비즈니스 로직 (일부만 사용)
│   ├── api/routers/v1/     # FastAPI 라우터
│   └── worker/             # 백그라운드 태스크
└── utils/                  # 외부 API 클라이언트 (Suno, Creatomate, ChatGPT)

1.2 현재 문제점

문제	설명	영향
Fat Controller	라우터에 비즈니스 로직이 직접 포함됨	테스트 어려움, 재사용 불가
서비스 레이어 미활용	services/ 폴더가 있지만 대부분 사용되지 않음	코드 중복, 일관성 부족
외부 API 결합	라우터에서 직접 외부 API 호출	모킹 어려움, 의존성 강결합
Repository 부재	데이터 접근 로직이 분산됨	쿼리 중복, 최적화 어려움
트랜잭션 관리 분산	각 함수에서 개별적으로 세션 관리	일관성 부족
에러 처리 비일관	HTTPException이 여러 계층에서 발생	디버깅 어려움

1.3 현재 코드 예시 (문제점)

# app/lyric/api/routers/v1/lyric.py - 현재 구조
@router.post("/generate")
async def generate_lyric(
    request_body: GenerateLyricRequest,
    background_tasks: BackgroundTasks,
    session: AsyncSession = Depends(get_session),
) -> GenerateLyricResponse:
    task_id = request_body.task_id

    try:
        # 문제 1: 라우터에서 직접 비즈니스 로직 수행
        service = ChatgptService(
            customer_name=request_body.customer_name,
            region=request_body.region,
            ...
        )
        prompt = service.build_lyrics_prompt()

        # 문제 2: 라우터에서 직접 DB 조작
        project = Project(
            store_name=request_body.customer_name,
            ...
        )
        session.add(project)
        await session.commit()

        # 문제 3: 라우터에서 직접 모델 생성
        lyric = Lyric(
            project_id=project.id,
            ...
        )
        session.add(lyric)
        await session.commit()

        # 문제 4: 에러 처리가 각 함수마다 다름
        background_tasks.add_task(generate_lyric_background, ...)

        return GenerateLyricResponse(...)
    except Exception as e:
        await session.rollback()
        return GenerateLyricResponse(success=False, ...)

2. 제안하는 디자인 패턴

2.1 Clean Architecture + 레이어드 아키텍처

┌─────────────────────────────────────────────────────────────────┐
│                      Presentation Layer                         │
│  (FastAPI Routers - HTTP 요청/응답만 처리)                       │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                      Application Layer                          │
│  (Use Cases / Services - 비즈니스 로직 오케스트레이션)           │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                       Domain Layer                              │
│  (Entities, Value Objects, Domain Services)                     │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Infrastructure Layer                         │
│  (Repositories, External APIs, Database)                        │
└─────────────────────────────────────────────────────────────────┘

2.2 적용할 디자인 패턴

패턴	적용 대상	목적
Repository Pattern	데이터 접근	DB 로직 캡슐화, 테스트 용이
Service Pattern	비즈니스 로직	유스케이스 구현, 트랜잭션 관리
Factory Pattern	객체 생성	복잡한 객체 생성 캡슐화
Strategy Pattern	외부 API	API 클라이언트 교체 용이
Unit of Work	트랜잭션	일관된 트랜잭션 관리
Dependency Injection	전체	느슨한 결합, 테스트 용이
DTO Pattern	계층 간 전달	명확한 데이터 경계

3. 상세 리팩토링 방안

3.1 새로운 폴더 구조

app/
├── core/
│   ├── __init__.py
│   ├── config.py                    # 설정 관리 (기존 config.py 이동)
│   ├── exceptions.py                # 도메인 예외 정의
│   ├── interfaces/                  # 추상 인터페이스
│   │   ├── __init__.py
│   │   ├── repository.py            # IRepository 인터페이스
│   │   ├── service.py               # IService 인터페이스
│   │   └── external_api.py          # IExternalAPI 인터페이스
│   └── uow.py                       # Unit of Work
│
├── domain/
│   ├── __init__.py
│   ├── entities/                    # 도메인 엔티티
│   │   ├── __init__.py
│   │   ├── project.py
│   │   ├── lyric.py
│   │   ├── song.py
│   │   └── video.py
│   ├── value_objects/               # 값 객체
│   │   ├── __init__.py
│   │   ├── task_id.py
│   │   └── status.py
│   └── events/                      # 도메인 이벤트
│       ├── __init__.py
│       └── lyric_events.py
│
├── infrastructure/
│   ├── __init__.py
│   ├── database/
│   │   ├── __init__.py
│   │   ├── session.py               # DB 세션 관리
│   │   ├── models/                  # SQLAlchemy 모델
│   │   │   ├── __init__.py
│   │   │   ├── project_model.py
│   │   │   ├── lyric_model.py
│   │   │   ├── song_model.py
│   │   │   └── video_model.py
│   │   └── repositories/            # Repository 구현
│   │       ├── __init__.py
│   │       ├── base.py
│   │       ├── project_repository.py
│   │       ├── lyric_repository.py
│   │       ├── song_repository.py
│   │       └── video_repository.py
│   ├── external/                    # 외부 API 클라이언트
│   │   ├── __init__.py
│   │   ├── chatgpt/
│   │   │   ├── __init__.py
│   │   │   ├── client.py
│   │   │   └── prompts.py
│   │   ├── suno/
│   │   │   ├── __init__.py
│   │   │   └── client.py
│   │   ├── creatomate/
│   │   │   ├── __init__.py
│   │   │   └── client.py
│   │   └── azure_blob/
│   │       ├── __init__.py
│   │       └── client.py
│   └── cache/
│       ├── __init__.py
│       └── redis.py
│
├── application/
│   ├── __init__.py
│   ├── services/                    # 애플리케이션 서비스
│   │   ├── __init__.py
│   │   ├── lyric_service.py
│   │   ├── song_service.py
│   │   └── video_service.py
│   ├── dto/                         # Data Transfer Objects
│   │   ├── __init__.py
│   │   ├── lyric_dto.py
│   │   ├── song_dto.py
│   │   └── video_dto.py
│   └── tasks/                       # 백그라운드 태스크
│       ├── __init__.py
│       ├── lyric_tasks.py
│       ├── song_tasks.py
│       └── video_tasks.py
│
├── presentation/
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── lyric_router.py
│   │   │   ├── song_router.py
│   │   │   └── video_router.py
│   │   └── dependencies.py          # FastAPI 의존성
│   ├── schemas/                     # API 스키마 (요청/응답)
│   │   ├── __init__.py
│   │   ├── lyric_schema.py
│   │   ├── song_schema.py
│   │   └── video_schema.py
│   └── middleware/
│       ├── __init__.py
│       └── error_handler.py
│
└── main.py

3.2 Repository Pattern 구현

3.2.1 추상 인터페이스

# app/core/interfaces/repository.py
from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Optional, List

T = TypeVar("T")

class IRepository(ABC, Generic[T]):
    """Repository 인터페이스 - 데이터 접근 추상화"""

    @abstractmethod
    async def get_by_id(self, id: int) -> Optional[T]:
        """ID로 엔티티 조회"""
        pass

    @abstractmethod
    async def get_by_task_id(self, task_id: str) -> Optional[T]:
        """task_id로 엔티티 조회"""
        pass

    @abstractmethod
    async def get_all(
        self,
        skip: int = 0,
        limit: int = 100,
        filters: dict = None
    ) -> List[T]:
        """전체 엔티티 조회 (페이지네이션)"""
        pass

    @abstractmethod
    async def create(self, entity: T) -> T:
        """엔티티 생성"""
        pass

    @abstractmethod
    async def update(self, entity: T) -> T:
        """엔티티 수정"""
        pass

    @abstractmethod
    async def delete(self, id: int) -> bool:
        """엔티티 삭제"""
        pass

    @abstractmethod
    async def count(self, filters: dict = None) -> int:
        """엔티티 개수 조회"""
        pass

3.2.2 Base Repository 구현

# app/infrastructure/database/repositories/base.py
from typing import Generic, TypeVar, Optional, List, Type
from sqlalchemy import select, func, and_
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.interfaces.repository import IRepository
from app.infrastructure.database.session import Base

T = TypeVar("T", bound=Base)

class BaseRepository(IRepository[T], Generic[T]):
    """Repository 기본 구현"""

    def __init__(self, session: AsyncSession, model: Type[T]):
        self._session = session
        self._model = model

    async def get_by_id(self, id: int) -> Optional[T]:
        result = await self._session.execute(
            select(self._model).where(self._model.id == id)
        )
        return result.scalar_one_or_none()

    async def get_by_task_id(self, task_id: str) -> Optional[T]:
        result = await self._session.execute(
            select(self._model)
            .where(self._model.task_id == task_id)
            .order_by(self._model.created_at.desc())
            .limit(1)
        )
        return result.scalar_one_or_none()

    async def get_all(
        self,
        skip: int = 0,
        limit: int = 100,
        filters: dict = None
    ) -> List[T]:
        query = select(self._model)

        if filters:
            conditions = [
                getattr(self._model, key) == value
                for key, value in filters.items()
                if hasattr(self._model, key)
            ]
            if conditions:
                query = query.where(and_(*conditions))

        query = query.offset(skip).limit(limit).order_by(
            self._model.created_at.desc()
        )
        result = await self._session.execute(query)
        return list(result.scalars().all())

    async def create(self, entity: T) -> T:
        self._session.add(entity)
        await self._session.flush()
        await self._session.refresh(entity)
        return entity

    async def update(self, entity: T) -> T:
        await self._session.flush()
        await self._session.refresh(entity)
        return entity

    async def delete(self, id: int) -> bool:
        entity = await self.get_by_id(id)
        if entity:
            await self._session.delete(entity)
            return True
        return False

    async def count(self, filters: dict = None) -> int:
        query = select(func.count(self._model.id))

        if filters:
            conditions = [
                getattr(self._model, key) == value
                for key, value in filters.items()
                if hasattr(self._model, key)
            ]
            if conditions:
                query = query.where(and_(*conditions))

        result = await self._session.execute(query)
        return result.scalar() or 0

3.2.3 특화된 Repository

# app/infrastructure/database/repositories/lyric_repository.py
from typing import Optional, List
from sqlalchemy import select

from app.infrastructure.database.repositories.base import BaseRepository
from app.infrastructure.database.models.lyric_model import LyricModel

class LyricRepository(BaseRepository[LyricModel]):
    """Lyric 전용 Repository"""

    def __init__(self, session):
        super().__init__(session, LyricModel)

    async def get_by_project_id(self, project_id: int) -> List[LyricModel]:
        """프로젝트 ID로 가사 목록 조회"""
        result = await self._session.execute(
            select(self._model)
            .where(self._model.project_id == project_id)
            .order_by(self._model.created_at.desc())
        )
        return list(result.scalars().all())

    async def get_completed_lyrics(
        self,
        skip: int = 0,
        limit: int = 100
    ) -> List[LyricModel]:
        """완료된 가사만 조회"""
        return await self.get_all(
            skip=skip,
            limit=limit,
            filters={"status": "completed"}
        )

    async def update_status(
        self,
        task_id: str,
        status: str,
        result: Optional[str] = None
    ) -> Optional[LyricModel]:
        """가사 상태 업데이트"""
        lyric = await self.get_by_task_id(task_id)
        if lyric:
            lyric.status = status
            if result is not None:
                lyric.lyric_result = result
            return await self.update(lyric)
        return None

3.3 Unit of Work Pattern

# app/core/uow.py
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from app.infrastructure.database.repositories.project_repository import ProjectRepository
    from app.infrastructure.database.repositories.lyric_repository import LyricRepository
    from app.infrastructure.database.repositories.song_repository import SongRepository
    from app.infrastructure.database.repositories.video_repository import VideoRepository

class IUnitOfWork(ABC):
    """Unit of Work 인터페이스"""

    projects: "ProjectRepository"
    lyrics: "LyricRepository"
    songs: "SongRepository"
    videos: "VideoRepository"

    @abstractmethod
    async def __aenter__(self):
        pass

    @abstractmethod
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        pass

    @abstractmethod
    async def commit(self):
        pass

    @abstractmethod
    async def rollback(self):
        pass


# app/infrastructure/database/uow.py
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.uow import IUnitOfWork
from app.infrastructure.database.session import AsyncSessionLocal
from app.infrastructure.database.repositories.project_repository import ProjectRepository
from app.infrastructure.database.repositories.lyric_repository import LyricRepository
from app.infrastructure.database.repositories.song_repository import SongRepository
from app.infrastructure.database.repositories.video_repository import VideoRepository

class UnitOfWork(IUnitOfWork):
    """Unit of Work 구현 - 트랜잭션 관리"""

    def __init__(self, session_factory=AsyncSessionLocal):
        self._session_factory = session_factory
        self._session: AsyncSession = None

    async def __aenter__(self):
        self._session = self._session_factory()

        # Repository 인스턴스 생성
        self.projects = ProjectRepository(self._session)
        self.lyrics = LyricRepository(self._session)
        self.songs = SongRepository(self._session)
        self.videos = VideoRepository(self._session)

        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            await self.rollback()
        await self._session.close()

    async def commit(self):
        await self._session.commit()

    async def rollback(self):
        await self._session.rollback()

3.4 Service Layer 구현

# app/application/services/lyric_service.py
from typing import Optional
from dataclasses import dataclass

from app.core.uow import IUnitOfWork
from app.core.exceptions import (
    EntityNotFoundError,
    ExternalAPIError,
    ValidationError
)
from app.application.dto.lyric_dto import (
    CreateLyricDTO,
    LyricResponseDTO,
    LyricStatusDTO
)
from app.infrastructure.external.chatgpt.client import IChatGPTClient
from app.infrastructure.database.models.lyric_model import LyricModel
from app.infrastructure.database.models.project_model import ProjectModel

@dataclass
class LyricService:
    """Lyric 비즈니스 로직 서비스"""

    uow: IUnitOfWork
    chatgpt_client: IChatGPTClient

    async def create_lyric(self, dto: CreateLyricDTO) -> LyricResponseDTO:
        """가사 생성 요청 처리

        1. 프롬프트 생성
        2. Project 저장
        3. Lyric 저장 (processing)
        4. task_id 반환 (백그라운드 처리는 별도)
        """
        async with self.uow:
            # 프롬프트 생성
            prompt = self.chatgpt_client.build_lyrics_prompt(
                customer_name=dto.customer_name,
                region=dto.region,
                detail_region_info=dto.detail_region_info,
                language=dto.language
            )

            # Project 생성
            project = ProjectModel(
                store_name=dto.customer_name,
                region=dto.region,
                task_id=dto.task_id,
                detail_region_info=dto.detail_region_info,
                language=dto.language
            )
            project = await self.uow.projects.create(project)

            # Lyric 생성 (processing 상태)
            lyric = LyricModel(
                project_id=project.id,
                task_id=dto.task_id,
                status="processing",
                lyric_prompt=prompt,
                language=dto.language
            )
            lyric = await self.uow.lyrics.create(lyric)

            await self.uow.commit()

            return LyricResponseDTO(
                success=True,
                task_id=dto.task_id,
                lyric=None,  # 백그라운드에서 생성
                language=dto.language,
                prompt=prompt  # 백그라운드 태스크에 전달
            )

    async def process_lyric_generation(
        self,
        task_id: str,
        prompt: str,
        language: str
    ) -> None:
        """백그라운드에서 가사 실제 생성

        이 메서드는 백그라운드 태스크에서 호출됨
        """
        try:
            # ChatGPT로 가사 생성
            result = await self.chatgpt_client.generate(prompt)

            # 실패 패턴 검사
            is_failure = self._check_failure_patterns(result)

            async with self.uow:
                status = "failed" if is_failure else "completed"
                await self.uow.lyrics.update_status(
                    task_id=task_id,
                    status=status,
                    result=result
                )
                await self.uow.commit()

        except Exception as e:
            async with self.uow:
                await self.uow.lyrics.update_status(
                    task_id=task_id,
                    status="failed",
                    result=f"Error: {str(e)}"
                )
                await self.uow.commit()
            raise

    async def get_lyric_status(self, task_id: str) -> LyricStatusDTO:
        """가사 생성 상태 조회"""
        async with self.uow:
            lyric = await self.uow.lyrics.get_by_task_id(task_id)

            if not lyric:
                raise EntityNotFoundError(
                    f"task_id '{task_id}'에 해당하는 가사를 찾을 수 없습니다."
                )

            status_messages = {
                "processing": "가사 생성 중입니다.",
                "completed": "가사 생성이 완료되었습니다.",
                "failed": "가사 생성에 실패했습니다.",
            }

            return LyricStatusDTO(
                task_id=lyric.task_id,
                status=lyric.status,
                message=status_messages.get(lyric.status, "알 수 없는 상태입니다.")
            )

    async def get_lyric_detail(self, task_id: str) -> LyricResponseDTO:
        """가사 상세 조회"""
        async with self.uow:
            lyric = await self.uow.lyrics.get_by_task_id(task_id)

            if not lyric:
                raise EntityNotFoundError(
                    f"task_id '{task_id}'에 해당하는 가사를 찾을 수 없습니다."
                )

            return LyricResponseDTO(
                id=lyric.id,
                task_id=lyric.task_id,
                project_id=lyric.project_id,
                status=lyric.status,
                lyric_prompt=lyric.lyric_prompt,
                lyric_result=lyric.lyric_result,
                language=lyric.language,
                created_at=lyric.created_at
            )

    def _check_failure_patterns(self, result: str) -> bool:
        """ChatGPT 응답에서 실패 패턴 검사"""
        failure_patterns = [
            "ERROR:",
            "I'm sorry",
            "I cannot",
            "I can't",
            "I apologize",
            "I'm unable",
            "I am unable",
            "I'm not able",
            "I am not able",
        ]
        return any(
            pattern.lower() in result.lower()
            for pattern in failure_patterns
        )

3.5 DTO (Data Transfer Objects)

# app/application/dto/lyric_dto.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class CreateLyricDTO:
    """가사 생성 요청 DTO"""
    task_id: str
    customer_name: str
    region: str
    detail_region_info: Optional[str] = None
    language: str = "Korean"

@dataclass
class LyricResponseDTO:
    """가사 응답 DTO"""
    success: bool = True
    task_id: Optional[str] = None
    lyric: Optional[str] = None
    language: str = "Korean"
    error_message: Optional[str] = None

    # 상세 조회 시 추가 필드
    id: Optional[int] = None
    project_id: Optional[int] = None
    status: Optional[str] = None
    lyric_prompt: Optional[str] = None
    lyric_result: Optional[str] = None
    created_at: Optional[datetime] = None
    prompt: Optional[str] = None  # 백그라운드 태스크용

@dataclass
class LyricStatusDTO:
    """가사 상태 조회 DTO"""
    task_id: str
    status: str
    message: str

3.6 Strategy Pattern for External APIs

# app/core/interfaces/external_api.py
from abc import ABC, abstractmethod
from typing import Optional

class ILLMClient(ABC):
    """LLM 클라이언트 인터페이스"""

    @abstractmethod
    def build_lyrics_prompt(
        self,
        customer_name: str,
        region: str,
        detail_region_info: str,
        language: str
    ) -> str:
        pass

    @abstractmethod
    async def generate(self, prompt: str) -> str:
        pass


class IMusicGeneratorClient(ABC):
    """음악 생성 클라이언트 인터페이스"""

    @abstractmethod
    async def generate(
        self,
        prompt: str,
        genre: str,
        callback_url: Optional[str] = None
    ) -> str:
        """음악 생성 요청, task_id 반환"""
        pass

    @abstractmethod
    async def get_status(self, task_id: str) -> dict:
        pass


class IVideoGeneratorClient(ABC):
    """영상 생성 클라이언트 인터페이스"""

    @abstractmethod
    async def get_template(self, template_id: str) -> dict:
        pass

    @abstractmethod
    async def render(self, source: dict) -> dict:
        pass

    @abstractmethod
    async def get_render_status(self, render_id: str) -> dict:
        pass


# app/infrastructure/external/chatgpt/client.py
from openai import AsyncOpenAI

from app.core.interfaces.external_api import ILLMClient
from app.infrastructure.external.chatgpt.prompts import LYRICS_PROMPT_TEMPLATE

class ChatGPTClient(ILLMClient):
    """ChatGPT 클라이언트 구현"""

    def __init__(self, api_key: str, model: str = "gpt-5-mini"):
        self._client = AsyncOpenAI(api_key=api_key)
        self._model = model

    def build_lyrics_prompt(
        self,
        customer_name: str,
        region: str,
        detail_region_info: str,
        language: str
    ) -> str:
        return LYRICS_PROMPT_TEMPLATE.format(
            customer_name=customer_name,
            region=region,
            detail_region_info=detail_region_info,
            language=language
        )

    async def generate(self, prompt: str) -> str:
        completion = await self._client.chat.completions.create(
            model=self._model,
            messages=[{"role": "user", "content": prompt}]
        )
        return completion.choices[0].message.content or ""

3.7 Presentation Layer (Thin Router)

# app/presentation/api/v1/lyric_router.py
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status

from app.presentation.schemas.lyric_schema import (
    GenerateLyricRequest,
    GenerateLyricResponse,
    LyricStatusResponse,
    LyricDetailResponse
)
from app.application.services.lyric_service import LyricService
from app.application.dto.lyric_dto import CreateLyricDTO
from app.core.exceptions import EntityNotFoundError
from app.presentation.api.dependencies import get_lyric_service

router = APIRouter(prefix="/lyric", tags=["lyric"])

@router.post(
    "/generate",
    response_model=GenerateLyricResponse,
    summary="가사 생성",
    description="고객 정보를 기반으로 ChatGPT를 이용하여 가사를 생성합니다."
)
async def generate_lyric(
    request: GenerateLyricRequest,
    background_tasks: BackgroundTasks,
    service: LyricService = Depends(get_lyric_service)
) -> GenerateLyricResponse:
    """
    라우터는 HTTP 요청/응답만 처리
    비즈니스 로직은 서비스에 위임
    """
    # DTO로 변환
    dto = CreateLyricDTO(
        task_id=request.task_id,
        customer_name=request.customer_name,
        region=request.region,
        detail_region_info=request.detail_region_info,
        language=request.language
    )

    # 서비스 호출
    result = await service.create_lyric(dto)

    # 백그라운드 태스크 등록
    background_tasks.add_task(
        service.process_lyric_generation,
        task_id=result.task_id,
        prompt=result.prompt,
        language=result.language
    )

    # 응답 반환
    return GenerateLyricResponse(
        success=result.success,
        task_id=result.task_id,
        lyric=result.lyric,
        language=result.language,
        error_message=result.error_message
    )

@router.get(
    "/status/{task_id}",
    response_model=LyricStatusResponse,
    summary="가사 생성 상태 조회"
)
async def get_lyric_status(
    task_id: str,
    service: LyricService = Depends(get_lyric_service)
) -> LyricStatusResponse:
    try:
        result = await service.get_lyric_status(task_id)
        return LyricStatusResponse(
            task_id=result.task_id,
            status=result.status,
            message=result.message
        )
    except EntityNotFoundError as e:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(e)
        )

@router.get(
    "/{task_id}",
    response_model=LyricDetailResponse,
    summary="가사 상세 조회"
)
async def get_lyric_detail(
    task_id: str,
    service: LyricService = Depends(get_lyric_service)
) -> LyricDetailResponse:
    try:
        result = await service.get_lyric_detail(task_id)
        return LyricDetailResponse.model_validate(result.__dict__)
    except EntityNotFoundError as e:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(e)
        )

3.8 Dependency Injection 설정

# app/presentation/api/dependencies.py
from functools import lru_cache
from fastapi import Depends

from app.core.config import get_settings, Settings
from app.infrastructure.database.uow import UnitOfWork
from app.infrastructure.external.chatgpt.client import ChatGPTClient
from app.infrastructure.external.suno.client import SunoClient
from app.infrastructure.external.creatomate.client import CreatomateClient
from app.application.services.lyric_service import LyricService
from app.application.services.song_service import SongService
from app.application.services.video_service import VideoService

@lru_cache()
def get_settings() -> Settings:
    return Settings()

def get_chatgpt_client(
    settings: Settings = Depends(get_settings)
) -> ChatGPTClient:
    return ChatGPTClient(
        api_key=settings.CHATGPT_API_KEY,
        model="gpt-5-mini"
    )

def get_suno_client(
    settings: Settings = Depends(get_settings)
) -> SunoClient:
    return SunoClient(
        api_key=settings.SUNO_API_KEY,
        callback_url=settings.SUNO_CALLBACK_URL
    )

def get_creatomate_client(
    settings: Settings = Depends(get_settings)
) -> CreatomateClient:
    return CreatomateClient(
        api_key=settings.CREATOMATE_API_KEY
    )

def get_unit_of_work() -> UnitOfWork:
    return UnitOfWork()

def get_lyric_service(
    uow: UnitOfWork = Depends(get_unit_of_work),
    chatgpt: ChatGPTClient = Depends(get_chatgpt_client)
) -> LyricService:
    return LyricService(uow=uow, chatgpt_client=chatgpt)

def get_song_service(
    uow: UnitOfWork = Depends(get_unit_of_work),
    suno: SunoClient = Depends(get_suno_client)
) -> SongService:
    return SongService(uow=uow, suno_client=suno)

def get_video_service(
    uow: UnitOfWork = Depends(get_unit_of_work),
    creatomate: CreatomateClient = Depends(get_creatomate_client)
) -> VideoService:
    return VideoService(uow=uow, creatomate_client=creatomate)

3.9 도메인 예외 정의

# app/core/exceptions.py

class DomainException(Exception):
    """도메인 예외 기본 클래스"""

    def __init__(self, message: str, code: str = None):
        self.message = message
        self.code = code
        super().__init__(message)

class EntityNotFoundError(DomainException):
    """엔티티를 찾을 수 없음"""

    def __init__(self, message: str):
        super().__init__(message, code="ENTITY_NOT_FOUND")

class ValidationError(DomainException):
    """유효성 검증 실패"""

    def __init__(self, message: str):
        super().__init__(message, code="VALIDATION_ERROR")

class ExternalAPIError(DomainException):
    """외부 API 호출 실패"""

    def __init__(self, message: str, service: str = None):
        self.service = service
        super().__init__(message, code="EXTERNAL_API_ERROR")

class BusinessRuleViolation(DomainException):
    """비즈니스 규칙 위반"""

    def __init__(self, message: str):
        super().__init__(message, code="BUSINESS_RULE_VIOLATION")

3.10 전역 예외 핸들러

# app/presentation/middleware/error_handler.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from app.core.exceptions import (
    DomainException,
    EntityNotFoundError,
    ValidationError,
    ExternalAPIError
)

def setup_exception_handlers(app: FastAPI):
    """전역 예외 핸들러 설정"""

    @app.exception_handler(EntityNotFoundError)
    async def entity_not_found_handler(
        request: Request,
        exc: EntityNotFoundError
    ) -> JSONResponse:
        return JSONResponse(
            status_code=404,
            content={
                "success": False,
                "error": {
                    "code": exc.code,
                    "message": exc.message
                }
            }
        )

    @app.exception_handler(ValidationError)
    async def validation_error_handler(
        request: Request,
        exc: ValidationError
    ) -> JSONResponse:
        return JSONResponse(
            status_code=400,
            content={
                "success": False,
                "error": {
                    "code": exc.code,
                    "message": exc.message
                }
            }
        )

    @app.exception_handler(ExternalAPIError)
    async def external_api_error_handler(
        request: Request,
        exc: ExternalAPIError
    ) -> JSONResponse:
        return JSONResponse(
            status_code=503,
            content={
                "success": False,
                "error": {
                    "code": exc.code,
                    "message": exc.message,
                    "service": exc.service
                }
            }
        )

    @app.exception_handler(DomainException)
    async def domain_exception_handler(
        request: Request,
        exc: DomainException
    ) -> JSONResponse:
        return JSONResponse(
            status_code=500,
            content={
                "success": False,
                "error": {
                    "code": exc.code or "UNKNOWN_ERROR",
                    "message": exc.message
                }
            }
        )

4. 모듈별 구현 예시

4.1 Song 모듈 리팩토링

# app/application/services/song_service.py
from dataclasses import dataclass

from app.core.uow import IUnitOfWork
from app.core.interfaces.external_api import IMusicGeneratorClient
from app.application.dto.song_dto import CreateSongDTO, SongResponseDTO

@dataclass
class SongService:
    """Song 비즈니스 로직 서비스"""

    uow: IUnitOfWork
    suno_client: IMusicGeneratorClient

    async def create_song(self, dto: CreateSongDTO) -> SongResponseDTO:
        """음악 생성 요청"""
        async with self.uow:
            # Lyric 조회
            lyric = await self.uow.lyrics.get_by_task_id(dto.task_id)
            if not lyric:
                raise EntityNotFoundError(
                    f"task_id '{dto.task_id}'에 해당하는 가사를 찾을 수 없습니다."
                )

            # Song 생성
            song = SongModel(
                project_id=lyric.project_id,
                lyric_id=lyric.id,
                task_id=dto.task_id,
                status="processing",
                song_prompt=lyric.lyric_result,
                language=lyric.language
            )
            song = await self.uow.songs.create(song)

            # Suno API 호출
            suno_task_id = await self.suno_client.generate(
                prompt=lyric.lyric_result,
                genre=dto.genre,
                callback_url=dto.callback_url
            )

            # suno_task_id 업데이트
            song.suno_task_id = suno_task_id
            await self.uow.commit()

            return SongResponseDTO(
                success=True,
                task_id=dto.task_id,
                suno_task_id=suno_task_id
            )

    async def handle_callback(
        self,
        suno_task_id: str,
        audio_url: str,
        duration: float
    ) -> None:
        """Suno 콜백 처리"""
        async with self.uow:
            song = await self.uow.songs.get_by_suno_task_id(suno_task_id)
            if song:
                song.status = "completed"
                song.song_result_url = audio_url
                song.duration = duration
                await self.uow.commit()

4.2 Video 모듈 리팩토링

# app/application/services/video_service.py
from dataclasses import dataclass

from app.core.uow import IUnitOfWork
from app.core.interfaces.external_api import IVideoGeneratorClient
from app.application.dto.video_dto import CreateVideoDTO, VideoResponseDTO

@dataclass
class VideoService:
    """Video 비즈니스 로직 서비스"""

    uow: IUnitOfWork
    creatomate_client: IVideoGeneratorClient

    async def create_video(self, dto: CreateVideoDTO) -> VideoResponseDTO:
        """영상 생성 요청"""
        async with self.uow:
            # 관련 데이터 조회
            project = await self.uow.projects.get_by_task_id(dto.task_id)
            lyric = await self.uow.lyrics.get_by_task_id(dto.task_id)
            song = await self.uow.songs.get_by_task_id(dto.task_id)
            images = await self.uow.images.get_by_task_id(dto.task_id)

            # 유효성 검사
            self._validate_video_creation(project, lyric, song, images)

            # Video 생성
            video = VideoModel(
                project_id=project.id,
                lyric_id=lyric.id,
                song_id=song.id,
                task_id=dto.task_id,
                status="processing"
            )
            video = await self.uow.videos.create(video)
            await self.uow.commit()

        # 외부 API 호출 (트랜잭션 외부)
        try:
            render_id = await self._render_video(
                images=[img.img_url for img in images],
                lyrics=song.song_prompt,
                music_url=song.song_result_url,
                duration=song.duration,
                orientation=dto.orientation
            )

            # render_id 업데이트
            async with self.uow:
                video = await self.uow.videos.get_by_id(video.id)
                video.creatomate_render_id = render_id
                await self.uow.commit()

            return VideoResponseDTO(
                success=True,
                task_id=dto.task_id,
                creatomate_render_id=render_id
            )

        except Exception as e:
            async with self.uow:
                video = await self.uow.videos.get_by_id(video.id)
                video.status = "failed"
                await self.uow.commit()
            raise

    async def _render_video(
        self,
        images: list[str],
        lyrics: str,
        music_url: str,
        duration: float,
        orientation: str
    ) -> str:
        """Creatomate로 영상 렌더링"""
        # 템플릿 조회
        template_id = self._get_template_id(orientation)
        template = await self.creatomate_client.get_template(template_id)

        # 템플릿 수정
        modified_template = self._prepare_template(
            template, images, lyrics, music_url, duration
        )

        # 렌더링 요청
        result = await self.creatomate_client.render(modified_template)

        return result[0]["id"] if isinstance(result, list) else result["id"]

    def _validate_video_creation(self, project, lyric, song, images):
        """영상 생성 유효성 검사"""
        if not project:
            raise EntityNotFoundError("Project를 찾을 수 없습니다.")
        if not lyric:
            raise EntityNotFoundError("Lyric을 찾을 수 없습니다.")
        if not song:
            raise EntityNotFoundError("Song을 찾을 수 없습니다.")
        if not song.song_result_url:
            raise ValidationError("음악 URL이 없습니다.")
        if not images:
            raise EntityNotFoundError("이미지를 찾을 수 없습니다.")

5. 기대 효과

5.1 코드 품질 향상

측면	현재	개선 후	기대 효과
테스트 용이성	라우터에서 직접 DB/API 호출	Repository/Service 모킹 가능	단위 테스트 커버리지 80%+
코드 재사용	로직 중복	서비스 레이어 공유	중복 코드 50% 감소
유지보수	변경 시 여러 파일 수정	단일 책임 원칙	수정 범위 최소화
확장성	새 기능 추가 어려움	인터페이스 기반 확장	새 LLM/API 추가 용이

5.2 아키텍처 개선

변경 전:
Router → DB + External API (강결합)

변경 후:
Router → Service → Repository → DB
              ↓
         Interface → External API (약결합)

5.3 테스트 가능성

# 단위 테스트 예시
import pytest
from unittest.mock import AsyncMock, MagicMock

from app.application.services.lyric_service import LyricService
from app.application.dto.lyric_dto import CreateLyricDTO

@pytest.fixture
def mock_uow():
    uow = MagicMock()
    uow.__aenter__ = AsyncMock(return_value=uow)
    uow.__aexit__ = AsyncMock(return_value=None)
    uow.commit = AsyncMock()
    uow.lyrics = MagicMock()
    uow.projects = MagicMock()
    return uow

@pytest.fixture
def mock_chatgpt():
    client = MagicMock()
    client.build_lyrics_prompt = MagicMock(return_value="test prompt")
    client.generate = AsyncMock(return_value="생성된 가사")
    return client

@pytest.mark.asyncio
async def test_create_lyric_success(mock_uow, mock_chatgpt):
    # Given
    service = LyricService(uow=mock_uow, chatgpt_client=mock_chatgpt)
    dto = CreateLyricDTO(
        task_id="test-task-id",
        customer_name="테스트 업체",
        region="서울"
    )

    mock_uow.projects.create = AsyncMock(return_value=MagicMock(id=1))
    mock_uow.lyrics.create = AsyncMock(return_value=MagicMock(id=1))

    # When
    result = await service.create_lyric(dto)

    # Then
    assert result.success is True
    assert result.task_id == "test-task-id"
    mock_uow.commit.assert_called_once()

@pytest.mark.asyncio
async def test_get_lyric_status_not_found(mock_uow, mock_chatgpt):
    # Given
    service = LyricService(uow=mock_uow, chatgpt_client=mock_chatgpt)
    mock_uow.lyrics.get_by_task_id = AsyncMock(return_value=None)

    # When & Then
    with pytest.raises(EntityNotFoundError):
        await service.get_lyric_status("non-existent-id")

5.4 개발 생산성

항목	기대 개선
새 기능 개발	템플릿 기반으로 30% 단축
버그 수정	단일 책임으로 원인 파악 용이
코드 리뷰	계층별 리뷰로 효율성 향상
온보딩	명확한 구조로 학습 시간 단축

5.5 운영 안정성

항목	현재	개선 후
트랜잭션 관리	분산되어 일관성 부족	UoW로 일관된 관리
에러 처리	HTTPException 혼재	도메인 예외로 통일
로깅	각 함수에서 개별	서비스 레벨에서 일관
모니터링	어려움	서비스 경계에서 명확한 메트릭

6. 마이그레이션 전략

6.1 단계별 접근

Phase 1: 기반 구축 (1주)
├── core/ 인터페이스 정의
├── 도메인 예외 정의
└── Base Repository 구현

Phase 2: Lyric 모듈 리팩토링 (1주)
├── LyricRepository 구현
├── LyricService 구현
├── 라우터 슬림화
└── 테스트 작성

Phase 3: Song 모듈 리팩토링 (1주)
├── SongRepository 구현
├── SongService 구현
├── Suno 클라이언트 인터페이스화
└── 테스트 작성

Phase 4: Video 모듈 리팩토링 (1주)
├── VideoRepository 구현
├── VideoService 구현
├── Creatomate 클라이언트 인터페이스화
└── 테스트 작성

Phase 5: 정리 및 최적화 (1주)
├── 기존 코드 제거
├── 문서화
├── 성능 테스트
└── 리뷰 및 배포

6.2 점진적 마이그레이션 전략

기존 코드를 유지하면서 새 구조로 점진적 이전:

# 1단계: 새 서비스를 기존 라우터에서 호출
@router.post("/generate")
async def generate_lyric(
    request: GenerateLyricRequest,
    background_tasks: BackgroundTasks,
    session: AsyncSession = Depends(get_session),
    # 새 서비스 주입 (optional)
    service: LyricService = Depends(get_lyric_service)
) -> GenerateLyricResponse:
    # 피처 플래그로 분기
    if settings.USE_NEW_ARCHITECTURE:
        return await _generate_lyric_new(request, background_tasks, service)
    else:
        return await _generate_lyric_legacy(request, background_tasks, session)

6.3 리스크 관리

리스크	완화 전략
기능 회귀	기존 테스트 유지, 새 테스트 추가
성능 저하	벤치마크 테스트
배포 실패	피처 플래그로 롤백 가능
학습 곡선	문서화 및 페어 프로그래밍

결론

이 리팩토링을 통해:

명확한 책임 분리: 각 계층이 하나의 역할만 수행
높은 테스트 커버리지: 비즈니스 로직 단위 테스트 가능
유연한 확장성: 새로운 LLM/API 추가 시 인터페이스만 구현
일관된 에러 처리: 도메인 예외로 통일된 에러 응답
트랜잭션 안정성: Unit of Work로 데이터 일관성 보장

현재 프로젝트가 잘 동작하고 있다면, 점진적 마이그레이션을 통해 리스크를 최소화하면서 아키텍처를 개선할 수 있습니다.

작성일: 2024-12-29 버전: 1.0

48 KiB Raw Blame History

디자인 패턴 기반 리팩토링 제안서

목차

1. 현재 아키텍처 분석

1.1 현재 구조

1.2 현재 문제점

1.3 현재 코드 예시 (문제점)

2. 제안하는 디자인 패턴

2.1 Clean Architecture + 레이어드 아키텍처

2.2 적용할 디자인 패턴

3. 상세 리팩토링 방안

3.1 새로운 폴더 구조

3.2 Repository Pattern 구현

3.2.1 추상 인터페이스

3.2.2 Base Repository 구현

3.2.3 특화된 Repository

3.3 Unit of Work Pattern

3.4 Service Layer 구현

3.5 DTO (Data Transfer Objects)

3.6 Strategy Pattern for External APIs

3.7 Presentation Layer (Thin Router)

3.8 Dependency Injection 설정

3.9 도메인 예외 정의

3.10 전역 예외 핸들러

4. 모듈별 구현 예시

4.1 Song 모듈 리팩토링

4.2 Video 모듈 리팩토링

5. 기대 효과

5.1 코드 품질 향상

5.2 아키텍처 개선

5.3 테스트 가능성

5.4 개발 생산성

5.5 운영 안정성

6. 마이그레이션 전략

6.1 단계별 접근

6.2 점진적 마이그레이션 전략

6.3 리스크 관리

결론

48 KiB

Raw Blame History