[Python] Pythanic 하게 횡단 관심사 분리

Python

[Python] Pythanic 하게 횡단 관심사 분리

sian han 2025. 11. 21. 13:33

Python 데코레이터로 서비스 레이어의 횡단 관심사 분리하기

참고: 이 글의 코드 예시는 실제 업무에서 진행한 리팩토링을 기반으로 작성되었지만, 회사의 비즈니스 로직과 도메인 정보는 모두 제거하고 일반적인 예시(음식 주문 시스템)로 변경했음. 핵심은 횡단 관심사를 어떻게 분리했는가이지, 구체적인 비즈니스 로직이 아니기 때문. 패턴과 접근 방식은 동일하게 유지했음.

현재 서비스 로직

FastAPI 사용중

/api/v2/order
├── service_base.py      # 모든 서비스가 상속받음
├── main/
│   ├── router.py
│   ├── service.py
│   └── dependencies.py
└── platform/
    ├── router.py
    ├── service.py
    ├── schemas.py
    └── dependencies.py
... 기타 도메인

폴더 구조는 도메인 기반이다. FastAPI 커뮤니티에서 많이 참조하는 fastapi-best-practices를 따랐음. 이 가이드는 기능별로 모듈을 나누는 것보다 도메인별로 나누는 구조를 권장하는데, 각 도메인이 독립적으로 router, service, schema를 가지면서 응집도가 높아지고 유지보수가 쉬워지기 때문임.

각 패키지에는 고유한 라우터, 스키마, 모델 등이 있음

router.py - 모든 엔드포인트를 갖춘 각 모듈의 핵심임
schemas.py - Pydantic 모델용
service.py - 모듈별 비즈니스 로직
dependencies.py - 모듈별 상수 및 오류 코드
config.py - 환경변수
utils.py - 비즈니스 로직이 아닌 기능
exceptions.py - 모듈별 예외

배경: 대화형 주문 시스템 요구사항

버튼 기반 주문 시스템에서 요구사항은 아래와 같았음

1. 모든 대화 내용은 다시 조회 가능해야 함

언제든지 과거의 모든 대화 내용을 순서대로 다시 불러와 사용자에게 보여줄 수 있어야 함
= 모든 대화 history 는 DB 에 저장되어야 함

2. 대화 상태 유지

사용자가 대화 도중 페이지를 나가더라도, 이전에 진행했던 대화의 컨텍스트가 유지되어야 함(기간없음)

예시:

음식을 주문하시겠어요? > 예
어떤 카테고리를 선택하시겠어요? > 치킨

여기까지 선택하고 접속을 종료함. 30일 후 다시 접속했을 때, 처음부터 '음식을 주문하시겠어요?' 가 아닌, 그 다음 단계인 '어떤 치킨을 주문하시겠어요?' 부터 대화를 재개해야 함

3. 엄격한 접근제어

접근 권한이 없는 사용자의 데이터나 서비스에는 접근하면 안된다

문제 상황

이런 요구사항이 있다보니 아래와 같이 모든 서비스에서 반복되는 로직이 있음

보안 관련해서 검증하는 것은 router 레벨에서 처리하고 있음.
이건 의존성 주입으로 router 에서 처리할 수 있도록 해서 코드 간소화를 했음.

@router.post(
    "",  # /api/v2/order/platform
    response_model=SuccessResponse[common_schemas.OrderResponse],
    status_code=status.HTTP_200_OK,
    summary="배달 플랫폼 선택",
    description="사용자가 배달 플랫폼을 선택하는 단계",
    tags=["platform"]
)
async def select_platform(
    request: schemas.PlatformRequest,
    user_id: int = Depends(validate_token),
    _: None = Depends(validate_platform_access),
    order_service: service.OrderPlatformService = Depends(get_order_platform_service),
    db: AsyncSession = Depends(get_db)
) -> SuccessResponse[common_schemas.OrderResponse]:

그런데 문제는 Service 임

리팩토링 전 서비스 코드를 보면 문제가 명확하게 드러남.

아래 2개 서비스(OrderCategoryService, OrderPlatformService)를 보면 공통적으로 1. SessionContext 존재 확인, 2. session_states 업데이트, 3. 히스토리 저장 + 응답 반환 이 반복되고 있음.

서비스에서 메인 비즈니스 로직에 집중할 수 없음.

class OrderCategoryService(ServiceBase):
    """
    주문 카테고리 선택 서비스

    음식 카테고리 선택 기능 담당.
    ServiceBase를 상속받아 히스토리 자동 저장 기능을 사용합니다.
    """

    async def select_category(
        self,
        request: schemas.CategoryRequest,
        user_id: int
    ) -> Dict[str, Any]:
        """
        카테고리 선택 처리 (메인 함수)

        처리 순서:
        1. SessionContext 존재 확인
        2. session_states 업데이트 (CATEGORY, STEP_ID=3)
        3. 메뉴 선택 단계(STEP_ID=3) 봇 메시지 조회
        4. 히스토리 저장 및 응답 반환

        Args:
            request: CategoryRequest
            user_id: JWT에서 검증된 사용자 ID

        Returns:
            OrderResponse dict

        Raises:
            HTTPException: 검증 실패 시
        """
        try:
            # 1. SessionContext 존재 확인
            session = await _get_session_context(
                self.db,
                user_id,
                request.session_id
            )
            session_id = session.SESSION_ID

            # 2. session_states 업데이트
            await _update_category_state(
                self.db,
                session_id,
                request.category
            )

            # 3. 메뉴 선택 단계(STEP_ID=3) 봇 메시지 조회
            bot_messages, options = await _get_menu_selection_messages(self.db)

            # 4. 사용자 메시지 생성
            user_message = _build_category_user_message(request.category)

            # 5. 히스토리 저장 + 응답 반환 (템플릿 메서드 패턴)
            response = await self.execute_with_history(
                session_id=session_id,
                step_id=StepId.MENU_SELECTION.value,
                bot_messages=bot_messages,
                options=options,
                user_message=user_message
            )

            return response

        except HTTPException:
            raise
        except Exception as e:
            logger.error(
                f"카테고리 선택 실패: user_id={user_id}, "
                f"request={request.model_dump()}, error={str(e)}",
                exc_info=True
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"카테고리 선택 중 오류가 발생했습니다: {str(e)}"
            )

class OrderPlatformService(ServiceBase):
    """
    주문 플랫폼 서비스

    배달 플랫폼 선택 기능 담당.
    ServiceBase를 상속받아 히스토리 자동 저장 기능을 사용합니다.
    """

    async def select_platform(
        self,
        request: schemas.PlatformRequest,
        user_id: int
    ) -> Dict[str, Any]:
        """
        배달 플랫폼 선택 처리 (메인 함수)

        처리 순서:
        1. SessionContext 존재 확인
        2. 플랫폼별 사용자 정보 존재 여부 검증
        3. session_states 업데이트 (PLATFORM_TYPE, STEP_ID=2)
        4. 카테고리 선택 단계(STEP_ID=2) 봇 메시지 조회
        5. 히스토리 저장 및 응답 반환

        Args:
            request: PlatformRequest
            user_id: JWT에서 검증된 사용자 ID

        Returns:
            OrderResponse dict

        Raises:
            HTTPException: 검증 실패 시
        """
        try:
            # 1. SessionContext 존재 확인
            session = await _get_session_context(
                self.db,
                user_id,
                request.session_id
            )
            session_id = session.SESSION_ID

            # 2. 플랫폼별 사용자 정보 존재 여부 검증
            await _validate_platform_user_info(
                self.db,
                request.platform_type,
                request.session_id,
                request.delivery_address
            )

            # 3. session_states 업데이트
            await _update_platform_state(
                self.db,
                session_id,
                request.platform_type
            )

            # 4. 카테고리 선택 단계(STEP_ID=2) 봇 메시지 조회
            bot_messages, options = await _get_category_selection_messages(self.db)

            # 5. 사용자 메시지 생성
            user_message = _build_platform_selection_user_message(
                request.platform_type,
                request.delivery_address
            )

            # 6. 히스토리 저장 + 응답 반환 (템플릿 메서드 패턴)
            response = await self.execute_with_history(
                session_id=session_id,
                step_id=StepId.CATEGORY_SELECTION.value,
                bot_messages=bot_messages,
                options=options,
                user_message=user_message
            )

            return response

        except HTTPException:
            raise
        except Exception as e:
            logger.error(
                f"플랫폼 선택 실패: user_id={user_id}, "
                f"request={request.model_dump()}, error={str(e)}",
                exc_info=True
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"플랫폼 선택 중 오류가 발생했습니다: {str(e)}"
            )

해결: 데코레이터를 사용하여 횡단 관심사 분리

반복되는 횡단 관심사를 데코레이터로 추출함. order/decorators.py 파일을 만들어 @flow_handler 데코레이터를 정의함

def flow_handler(operation_name: str):
    """
    대화형 플로우 공통 처리 데코레이터

    서비스 메서드에 적용하여 반복적인 횡단 관심사를 자동으로 처리합니다.

    자동으로 처리하는 작업:
    1. SessionContext 존재 확인 및 session_id 추출
    2. 핵심 비즈니스 로직 실행 (session_id를 키워드 인자로 주입)
    3. session_states 업데이트
    4. 히스토리 저장 및 응답 반환
    5. 에러 핸들링 (HTTPException 재발생, Exception은 로깅 후 500 에러)

    Args:
        operation_name: 작업 이름 (로깅 및 에러 메시지에 사용)

    Returns:
        데코레이터 함수

    Note:
        - 데코레이트된 메서드는 반드시 FlowResult를 반환해야 합니다.
        - session_id는 데코레이터가 자동으로 주입하므로 메서드 시그니처에 포함해야 합니다.
        - request 객체는 반드시 session_id 속성을 가져야 합니다.
    """
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(
            self,
            request,
            user_id: int
        ) -> Dict[str, Any]:
            try:
                # 1. SessionContext 존재 확인 및 session_id 추출
                session = await get_session_context(
                    self.db,
                    user_id,
                    request.session_id
                )
                session_id = session.SESSION_ID

                # 2. 핵심 비즈니스 로직 실행 (session_id를 키워드 인자로 주입)
                result: FlowResult = await func(
                    self,
                    request,
                    user_id,
                    session_id=session_id
                )

                # 3. session_states 업데이트
                await update_session_state(
                    self.db,
                    session_id,
                    STEP_ID=result.next_step_id,
                    **result.state_updates  # 동적으로 필드 업데이트
                )

                # 4. 히스토리 저장 + 응답 반환 (템플릿 메서드 패턴)
                response = await self.execute_with_history(
                    session_id=session_id,
                    step_id=result.next_step_id,
                    bot_messages=result.bot_messages,
                    options=result.options,
                    user_message=result.user_message
                )

                return response

            except HTTPException:
                # HTTPException은 그대로 재발생 (이미 적절한 상태 코드와 메시지 포함)
                raise
            except Exception as e:
                # 일반 예외는 로깅 후 500 에러로 변환
                logger.error(
                    f"{operation_name} 실패: user_id={user_id}, "
                    f"request={request.model_dump()}, error={str(e)}",
                    exc_info=True
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail=f"{operation_name} 중 오류가 발생했습니다: {str(e)}"
                )

        return wrapper
    return decorator

데코레이터 적용 후 서비스 코드

@flow_handler 데코레이터를 적용하면 서비스 메서드는 핵심 비즈니스 로직만 남게 됨

리팩토링 전 약 60줄이었던 코드가 약 30줄로 줄어듦 (50% 감소)

class OrderPlatformService(ServiceBase):
    """
    주문 플랫폼 서비스

    배달 플랫폼 선택 기능 담당.
    ServiceBase를 상속받아 히스토리 자동 저장 기능을 사용합니다.
    """

    @flow_handler("플랫폼 선택")
    async def select_platform(
        self,
        request: schemas.PlatformRequest,
        user_id: int,
        session_id: int  # 데코레이터가 자동으로 주입
    ) -> FlowResult:
        """
        배달 플랫폼 선택 처리

        핵심 비즈니스 로직:
        1. 플랫폼별 사용자 정보 존재 여부 검증
        2. 카테고리 선택 단계(STEP_ID=2) 봇 메시지 조회
        3. 사용자 메시지 생성
        4. FlowResult 반환

        Args:
            request: PlatformRequest
            user_id: JWT에서 검증된 사용자 ID
            session_id: 세션 ID (데코레이터가 주입)

        Returns:
            FlowResult (데코레이터가 최종 응답으로 변환)
        """
        # 1. 플랫폼별 사용자 정보 존재 여부 검증 (도메인 특화 로직)
        await _validate_platform_user_info(
            self.db,
            request.platform_type,
            request.session_id,
            request.delivery_address
        )

        # 2. 카테고리 선택 단계(STEP_ID=2) 봇 메시지 조회
        bot_messages, options = await _get_category_selection_messages(self.db)

        # 3. 사용자 메시지 생성
        user_message = _build_platform_selection_user_message(
            request.platform_type,
            request.delivery_address
        )

        # 4. FlowResult 반환 (데코레이터가 나머지 자동 처리)
        return FlowResult(
            bot_messages=bot_messages,
            options=options,
            user_message=user_message,
            state_updates={'PLATFORM_TYPE': request.platform_type},
            next_step_id=StepId.CATEGORY_SELECTION.value
        )

개선 효과

1. 코드 라인 수 감소

Before: 약 60줄
After: 약 30줄
50% 감소

2. 비즈니스 로직에 집중 가능

서비스 메서드를 보면 이제 진짜 중요한 것만 보임

플랫폼 검증
메시지 조회
사용자 메시지 생성
결과 반환

횡단 관심사(세션 체크, 상태 업데이트, 히스토리 저장, 에러 핸들링)는 데코레이터가 알아서 처리

3. 유지보수성 향상

횡단 관심사 로직 수정이 필요하면 데코레이터 한 곳만 수정하면 됨
모든 서비스에 자동으로 반영됨

4. 테스트 용이성

비즈니스 로직만 독립적으로 테스트 가능
데코레이터도 별도로 테스트 가능

배운 점

횡단 관심사는 과감하게 분리하자
처음엔 "이 정도는 괜찮아"라고 생각했는데, 서비스가 늘어날수록 반복 코드가 기하급수적으로 늘어남
데코레이터는 강력하다
Python의 데코레이터는 이런 상황에 정말 잘 맞음. 코드를 수정하지 않고도 기능을 추가할 수 있음
명확한 책임 분리
- Router: 요청 검증, 보안
- Decorator: 횡단 관심사
- Service: 핵심 비즈니스 로직
이렇게 분리하니까 각각의 역할이 명확해짐
리팩토링은 점진적으로
처음부터 완벽한 구조를 만들려고 하지 말고, 반복이 보이면 그때 리팩토링하는게 더 실용적임

저작자표시 (새창열림)

'Python' 카테고리의 다른 글

[Python] 모듈 (0)	2025.09.25
[Python] 자료구조 (0)	2025.09.25
[Python] 조건문·반복문·함수 (0)	2025.09.25

현재글[Python] Pythanic 하게 횡단 관심사 분리

일	월	화	수	목	금	토
				1	2	3
4	5	6	7	8	9	10
11	12	13	14	15	16	17
18	19	20	21	22	23	24
25	26	27	28	29	30	31

seohyun han