이코에코(Eco²) MQ 도입 전, 코드 품질 개선: Character API 리팩토링

 
Character API는 재활용 스캔 리워드 시스템의 핵심 서비스입니다.
사용자가 재활용품을 스캔하면 해당 분류에 맞는 캐릭터를 리워드로 지급합니다.
캐릭터를 결과와 매칭, 지급, 관리, 기록하는 도메인이며 Scan(폐기물 이미지 분석), My(유저 정보)와 연동됩니다.
Character API에서 진행한 품질 개선 기준과 절차를 전 도메인에 적용합니다. (ext-authz[Go] 제외, 점검 결과 코드 품질 우수)

리워드 지급 흐름

 

리팩토링 전 문제점

• Race Condition: 동시 요청 시 중복 캐릭터 지급
• Dead Code: 미사용 메서드 및 예외 핸들러
• 하드코딩: 일부 설정값이 코드에 직접 기입
• 테스트 부족: 단위 테스트 미비
• Observability 부재: gRPC 분산 추적 미지원

---

1차 개선 (P0-P5)

우선순위	이슈	해결 방법
P0	Race Condition	DB UniqueConstraint + IntegrityError 핸들링
P1	Dead Code	Strategy 패턴으로 평가 로직 분리
P2	하드코딩 상수	core/constants.py 분리
P3	불완전한 타입 힌트	전체 타입 정리
P4	테스트 부족	단위 테스트 추가
P5	Circuit Breaker 미적용	aiobreaker 기반 구현

P0: Race Condition 해결

문제: 동시에 같은 캐릭터 지급 요청 시 중복 저장
해결: Optimistic Locking + IntegrityError 처리

# domains/character/services/character.py
async def _apply_reward(self, user_id: UUID, matches: list[Character], source: str):
    for character in matches:
        try:
            await self._grant_and_sync(user_id, character, source)
            return character, False  # 성공
        except IntegrityError:
            await self.session.rollback()
            return character, True  # 이미 소유

P5: Circuit Breaker 적용

문제: My 도메인 gRPC 호출 실패 시 연쇄 장애
해결: aiobreaker 기반 Circuit Breaker

# domains/character/rpc/my_client.py
class MyUserCharacterClient:
    def __init__(self, settings: Settings):
        self._circuit_breaker = CircuitBreaker(
            name="my-grpc-client",
            fail_max=settings.circuit_fail_max,        # 5회
            timeout_duration=settings.circuit_timeout_duration,  # 30초
        )

---

2차 개선 (코드 품질 심층)

우선순위	이슈	해결 방법
P0	warmup_catalog_cache 리소스 누수	finally 블록에서 engine.dispose()
P1	Circuit Breaker 설정 하드코딩	Settings에서 주입
P1	EvaluationContext 책임 과다	순수 함수 기반으로 변경
P2	Registry 모듈 로드 side effect	Lazy Initialization
P2	테스트 fixture __new__ 사용	Factory Method 패턴
P3	should_evaluate async 불필요	sync로 변경
P3	grant_character 인자 과다 (7개)	DTO 캡슐화

P1: EvaluationContext 책임 분리

Before: Evaluator가 DB 접근

# AS-IS
class ScanRewardEvaluator:
    async def match_characters(self, ctx: EvaluationContext):
        return await ctx.character_repo.find_by_match_label(...)

After: Service에서 데이터 전달, Evaluator는 순수 함수

# TO-BE
class ScanRewardEvaluator:
    def match_characters(self, payload, characters: list[Character]):
        return [c for c in characters if c.match_label == match_label]

P3: DTO 캡슐화

Before: 7개 인자

await client.grant_character(
    user_id, character_id, code, name, type, dialog, source
)

After: DTO로 캡슐화

@dataclass
class GrantCharacterRequest:
    user_id: UUID
    character_id: UUID
    character_code: str
    character_name: str
    character_type: str | None
    character_dialog: str | None
    source: str

await client.grant_character(request)

---

3차 개선 (복잡도 개선)

Radon 분석 결과 C등급(복잡도 11-20) 함수 7개를 모두 개선.

함수	파일	개선 방법
_record_reward_metrics	character.py	_determine_reward_status 분리
upsert_from_profile	user_repository.py	_resolve_or_create_user, _link_social_account, _sync_user_profile 분리
_build_reward_request	scan.py	_extract_classification 공통 헬퍼 추출
warmup_catalog_cache	cache.py	_load_and_cache_catalog, _character_to_profile 분리
load_catalog	import_character_catalog.py	_parse_catalog_row 분리
_get_request_origin	auth.py	_resolve_host, _resolve_scheme, _first_value 분리
resolve_csv_path	_csv_utils.py	_build_candidates, _find_by_keywords 분리

---

아키텍처 패턴

Strategy Pattern

리워드 평가 로직을 소스별로 분리하여 확장성 확보.

services/evaluators/
├── base.py          # RewardEvaluator 추상 클래스
├── registry.py      # Evaluator 등록/조회
└── scan.py          # ScanRewardEvaluator

Circuit Breaker Pattern

외부 서비스(My 도메인) 장애 시 fail-fast로 시스템 안정성 확보.

상태 전이:
CLOSED → (5회 실패) → OPEN → (30초 후) → HALF_OPEN → (성공) → CLOSED

Cache-Aside Pattern

Redis 캐시 + Graceful Degradation.

async def catalog():
    cached = await get_cached(CATALOG_KEY)
    if cached:
        return cached

    data = await repo.list_all()
    await set_cached(CATALOG_KEY, data, ttl=300)
    return data

Factory Method Pattern

테스트 의존성 주입을 위한 팩토리 메서드.

class CharacterService:
    @classmethod
    def create_for_test(cls, session, character_repo=None, ownership_repo=None):
        service = cls.__new__(cls)
        service.session = session
        service.character_repo = character_repo or CharacterRepository(session)
        service.ownership_repo = ownership_repo or CharacterOwnershipRepository(session)
        return service

---

테스트 전략

테스트 구조

tests/
├── conftest.py              # 공통 fixture
├── test_character_service.py  # 서비스 단위 테스트
├── test_evaluators.py       # Strategy 패턴 테스트
├── test_my_client.py        # gRPC 클라이언트 테스트
├── test_cache.py            # 캐시 레이어 테스트
├── integration/             # testcontainers 기반
└── e2e/                     # API 엔드포인트 테스트

Mock 전략

• DB Session: AsyncMock으로 대체
• Repository: MagicMock으로 메서드 단위 mock
• gRPC Client: AsyncMock으로 응답 시뮬레이션
• Redis: MagicMock으로 캐시 동작 시뮬레이션

---

실측 데이터

Radon 복잡도 분석

실행: radon cc domains/ -a -s --total-average

결과:
- 총 블록: 1,004개
- 평균 복잡도: A (2.51)
- C등급: 0개 (개선 전 7개)
- D/F등급: 0개

테스트 커버리지

실행: pytest domains/character/tests/ --cov=domains/character/services

결과:
- services/character.py: 91%
- services/evaluators/base.py: 100%
- services/evaluators/scan.py: 97%
- services/evaluators/registry.py: 83%
- 전체 services/: 92%

테스트 수

단위 테스트	0개	64개
통합 테스트	0개	별도 구성
E2E 테스트	0개	별도 구성

---

주요 성과

1. 안정성 향상: Race Condition 해결, Circuit Breaker 적용
2. 유지보수성 개선: 복잡도 A등급, 패턴 적용
3. 테스트 커버리지: 0% → 92% (서비스 레이어)
4. Observability: Jaeger 분산 추적 커버리지 확장 (Character gRPC <-> My gRPC)

---

Reference

• aiobreaker - Circuit Breaker
• Radon - Code Metrics
• pytest-asyncio

GitHub

• https://github.com/eco2-team/backend

Service

• 이코에코(Eco²) 분산 클러스터 실시간 배포 중