コーディング — レコメンドAPI バッチ処理クライアント改善

概要

問題

以下は「商品レコメンドAPIのバッチ処理クライアント」の実装です。 このコードには複数の設計上の問題があります。問題点を特定し、改善してください。

制約・前提条件

• Python 3.12 + httpx（非同期HTTPクライアント）を使用可能
• 1回の処理で最大 500 ユーザーを処理する
• APIは /recommend/batch エンドポイント（POST）で最大50件のuser_idを一括送信できる
• ユーザーへの返却型は型安全であること（Pydantic v2 使用）
• リトライ機能（最大3回、指数バックオフ）を組み込むこと

悪いコード (Before)

📋 コピー# bad_code.py
import requests
import json
import time

BASE_URL = "https://recommend-api.example.com"

def get_recommendations(user_ids):
    results = {}
    for user_id in user_ids:
        try:
            resp = requests.get(
                BASE_URL + "/recommend?user_id=" + str(user_id),
                timeout=10
            )
            if resp.status_code == 200:
                data = json.loads(resp.text)
                results[user_id] = data["items"]
            else:
                results[user_id] = []
        except:
            results[user_id] = []
        time.sleep(0.1)
    return results

def process_recommendations(user_ids):
    recs = get_recommendations(user_ids)
    output = []
    for uid, items in recs.items():
        for item in items:
            output.append({
                "user_id": uid,
                "item_id": item["id"],
                "score": item["score"],
            })
    return output

if __name__ == "__main__":
    users = [1001, 1002, 1003, 1004, 1005]
    result = process_recommendations(users)
    print(result)

ヒント（段階的開示）

import asyncio
import httpx
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_exponential

class RecommendItem(BaseModel):
    id: int
    score: float

class UserRecommendation(BaseModel):
    user_id: int
    items: list[RecommendItem]

def chunk(lst: list, size: int):
    for i in range(0, len(lst), size):
        yield lst[i:i+size]

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
async def fetch_batch(client: httpx.AsyncClient, user_ids: list[int]):
    resp = await client.post("/recommend/batch", json={"user_ids": user_ids})
    resp.raise_for_status()
    return [UserRecommendation.model_validate(item) for item in resp.json()]

async def get_recommendations(user_ids: list[int]):
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=10) as client:
        tasks = [fetch_batch(client, batch) for batch in chunk(user_ids, 50)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        # エラー処理...

問題点分析

バッチ並列処理フロー — Before vs After

模範解答

📋 コピー# good_code.py
import asyncio
from collections.abc import Generator

import httpx
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_exponential

BASE_URL = "https://recommend-api.example.com"
BATCH_SIZE = 50
REQUEST_TIMEOUT = 10.0


class RecommendItem(BaseModel):
    id: int
    score: float


class UserRecommendation(BaseModel):
    user_id: int
    items: list[RecommendItem]


class RecommendOutput(BaseModel):
    user_id: int
    item_id: int
    score: float


def _chunk(lst: list[int], size: int) -> Generator[list[int], None, None]:
    """リストを指定サイズに分割するジェネレータ。"""
    for i in range(0, len(lst), size):
        yield lst[i : i + size]


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=8),
    reraise=True,
)
async def _fetch_batch(
    client: httpx.AsyncClient,
    user_ids: list[int],
) -> list[UserRecommendation]:
    """バッチAPIを叩いてユーザーごとのレコメンド結果を返す（最大3回リトライ）。"""
    resp = await client.post("/recommend/batch", json={"user_ids": user_ids})
    resp.raise_for_status()
    return [UserRecommendation.model_validate(item) for item in resp.json()]


async def get_recommendations(
    user_ids: list[int],
) -> list[UserRecommendation]:
    """500件以下の user_ids を50件バッチで並列処理してレコメンドを取得する。"""
    batches = list(_chunk(user_ids, BATCH_SIZE))

    async with httpx.AsyncClient(
        base_url=BASE_URL,
        timeout=REQUEST_TIMEOUT,
    ) as client:
        tasks = [_fetch_batch(client, batch) for batch in batches]
        batch_results = await asyncio.gather(*tasks, return_exceptions=True)

    recommendations: list[UserRecommendation] = []
    for batch_ids, result in zip(batches, batch_results):
        if isinstance(result, Exception):
            # 失敗したバッチはログに記録して空リストで代替（部分的成功を優先）
            print(f"Batch failed for user_ids={batch_ids}: {result!r}")
        else:
            recommendations.extend(result)

    return recommendations


def process_recommendations(
    recommendations: list[UserRecommendation],
) -> list[RecommendOutput]:
    """UserRecommendation リストをフラットな RecommendOutput リストに変換する。"""
    return [
        RecommendOutput(user_id=rec.user_id, item_id=item.id, score=item.score)
        for rec in recommendations
        for item in rec.items
    ]


async def main() -> None:
    user_ids = list(range(1001, 1501))  # 500ユーザー
    recs = await get_recommendations(user_ids)
    outputs = process_recommendations(recs)
    print(f"総出力件数: {len(outputs)}")
    if outputs:
        print(f"先頭3件: {outputs[:3]}")


if __name__ == "__main__":
    asyncio.run(main())

ポイント解説

改善効果比較

実務への応用

• 販促メール配信前のレコメンド取得: 購読者リスト（数万件）に対してレコメンドAPIを呼び出すバッチ処理。 バッチ化 + 並列化で処理時間を劇的に短縮できる。
• Argo Workflows との統合: 並列処理は Argo の並列タスクとして実装するより、Python内のasyncioで完結させる方が管理しやすいケースもある。
• BigQuery コストとの比較視点: API呼び出し1回あたりのコストがある場合、バッチ化でコストを劇的に削減できる。

次のステップ

• 参考: httpx 公式ドキュメント（AsyncClient）
• 参考: tenacity ドキュメント
• 参考: Pydantic v2 model_validate

今日のまとめ

自己評価

自分の回答

気づき・メモ