[개발 기록] 대시보드 가격 API 응답 시간 줄이기 - 병렬 호출과 캐시 사용

1. 문제 상황

축산물 시세를 보여주는 대시보드에서 가격 조회가 5~10초 이상 걸리는 현상이 발생했습니다.

로그를 분석한 결과, KAMIS(농산물유통정보) API가 같은 부위에 대해 여러 번, 그리고 순차적으로 호출되고 있었으며, 이는 사용자에게 매우 느린 로딩 경험을 제공하고 있었습니다.

---

2. 원인 정리

2-1 등급별 API를 순차 3번 호출

"전체 등급"(grade_code=00)을 조회할 때, 국내 소고기는 1++, 1+, 1등급 데이터를 각각 따로 조회하여 평균을 내고 있었습니다. 즉, 부위 1개당 외부 API가 3번 연속 호출되었고, 각 호출이 끝난 뒤에야 다음 호출이 시작되는 구조였습니다.

기존 흐름: > [등급01 요청] → 응답 대기 → [등급02 요청] → 응답 대기 → [등급03 요청] → 응답 대기 → 평균 계산

2-2 부위별로도 순차 처리

대시보드에서 "등심 + 갈비" 두 부위를 조회할 때, 등심 조회가 끝난 뒤에 갈비를 조회했습니다. 등심 작업에만 3번의 API 호출이 필요하므로, 두 부위 조회 시 총 6번의 호출이 직렬로 실행되었습니다.

2-3 매 요청마다 API 먼저 호출

캐시는 "API 실패 시에만" 사용하고, 성공하는 경우에는 항상 KAMIS를 먼저 호출하는 구조였습니다. 같은 날 같은 부위를 여러 번 조회하더라도 매번 외부 API 요청이 발생했습니다.

---

3. 적용한 최적화 3가지

3-1 등급 01/02/03 병렬 조회 (asyncio.gather)

• 아이디어: 3개 등급을 동시에 요청하고, 결과만 모아서 평균을 낸다. 대기 시간은 "가장 오래 걸리는 1번 분량"으로 줄어듭니다.
• 변경 파일: meat_backend/apis.py — fetch_kamis_price

Before (순차):

grade_prices = []
for gc in ["01", "02", "03"]:
    price_data = await _fetch_kamis_price_single(..., grade_code=gc, ...)
    if price_data:
        grade_prices.append(...)

After (병렬):

async def _fetch_one_grade(gc: str):
    try:
        return await _fetch_kamis_price_single(
            part_name=part_name,
            region=region,
            grade_code=gc,
            target_day=target_day,
            key=key,
            cert_id=cert_id,
            base=base,
            codes=codes,
            county_code=county_code,
        )
    except Exception as e:
        logger.warning("등급 %s 조회 실패: %s", gc, e)
        return None

results = await asyncio.gather(
    *[_fetch_one_grade(gc) for gc in grade_codes_to_fetch],
    return_exceptions=False,
)

grade_prices = []
for gc, price_data in zip(grade_codes_to_fetch, results):
    if price_data:
        grade_prices.append({...})

• asyncio.gather로 3개 코루틴을 한 번에 실행합니다.
• return_exceptions=False 설정으로 예외는 전파되도록 하되, 개별 실패는 _fetch_one_grade 내부에서 None으로 처리하여 안정성을 높였습니다.

3-2. 부위별 가격 조회 병렬화 (대시보드)

• 아이디어: 소고기 부위들끼리, 돼지고기 부위들끼리 한꺼번에 조회합니다.
• 변경 파일: meat_backend/routes/dashboard.py — get_dashboard_prices

Before (순차):

for code, name in beef_parts:
    data = await price_service.fetch_current_price(...)
    if data: beef_items.append(...)

for code, name in pork_parts:
    data = await price_service.fetch_current_price(...)
    if data: pork_items.append(...)

After (병렬):

async def _fetch_beef(code: str, name: str):
    try:
        data = await price_service.fetch_current_price(
            part_name=code, region=region, grade_code=grade_code, db=db
        )
        if data.get("currentPrice", 0) > 0:
            return (code, name, data)
    except HTTPException as e:
        logger.warning(...)
    except Exception as e:
        logger.warning(...)
    return (code, name, None)

# _fetch_pork도 동일 패턴 적용

beef_results = await asyncio.gather(
    *[_fetch_beef(code, name) for code, name in beef_parts],
    return_exceptions=False,
)
pork_results = await asyncio.gather(
    *[_fetch_pork(code, name) for code, name in pork_parts],
    return_exceptions=False,
)

for _code, _name, data in beef_results:
    if data:
        beef_items.append(PriceItem(...))
for _code, _name, data in pork_results:
    if data:
        pork_items.append(PriceItem(...))

• 부위별로 (code, name, data) 튜플을 반환하게 하여, 실패 시 data=None인 경우만 걸러내도록 수정했습니다.
• 소/돼지 각각 gather 한 번으로 모든 부위를 동시에 조회합니다.

3-3 캐시 우선 (DB 먼저 조회)

• 아이디어: "API 먼저" 호출하는 대신 DB 캐시를 먼저 확인합니다. 당일 혹은 어제 데이터가 있다면 API를 호출하지 않고 즉시 반환합니다.
• 변경 파일: meat_backend/services/price_service.py — fetch_current_price

After (캐시 우선 전략):

today = date.today()
yesterday = today - timedelta(days=1)
cache_data: dict[str, Any] | None = None

# 1) 캐시 우선: DB에 최근(당일/어제) 데이터가 있으면 바로 반환
if db:
    cache_data = await self._get_from_db_cache(db, part_name, region, today)
    if cache_data:
        try:
            cache_date = datetime.strptime(cache_data["price_date"], "%Y-%m-%d").date()
            if cache_date >= yesterday:
                return {**cache_data, "source": "cache"}
        except (ValueError, TypeError):
            pass

# 2) 캐시 없거나 오래됨 → KAMIS API 호출
try:
    api_data = await self.kamis.fetch_current_price(...)
    if api_data.get("currentPrice", 0) > 0:
        if db:
            await self._save_to_db(...)
        return {**api_data, "source": "api"}
except HTTPException:
    raise
except Exception as e:
    logger.warning("KAMIS API call failed: %s", e)

# 3) API 실패 시 기존 캐시(오래된 것 포함) 반환
if db and cache_data:
    return {**cache_data, "source": "cache"}

raise HTTPException(...)

1. 첫 조회: 캐시 없음 → API 호출 → DB 저장 → 반환.
2. 이후 재조회: DB에 어제 이상의 데이터 존재 → API 호출 없이 즉시 반환.
3. 장애 대응: API 장애 시, 오래된 캐시 데이터라도 반환하여 사용자 경험을 유지합니다.

---

4. 부가 정리 (로그 최적화)

기존의 print("DEBUG: ...") 코드를 logger.debug(...)로 변경했습니다. 이를 통해 운영 환경에서는 로그 레벨을 조절하여 불필요한 I/O 부담을 줄일 수 있도록 개선했습니다.

---

5. 최종 정리

항목	Before	After
등급 3개 조회	순차 3회 (대기 3배)	병렬 1회 분량
부위 N개 조회	순차 N번	병렬 1번에 N개
재조회 시	매번 API 호출	DB 캐시 즉시 반환

• 첫 로딩: 병렬화 작업 덕분에 API 대기 시간이 획기적으로 줄었습니다.
• 재방문/새로고침: 캐시 우선 전략으로 대부분 DB만 조회하므로 응답 속도가 매우 빨라졌습니다.

핵심은 “기다려야 하는 일은 동시에 돌리고, 반복되는 조회는 캐시로 최소화하는 것”입니다.