【GCP】Cloud Run × Cloud Tasks で外部APIのレートリミットを回避する非同期バッチ構成（Python実装例つき）

[クライアント/API]
       │ (1. タスク追加リクエスト)
       ▼
[ Cloud Tasks (Queue) ] ──── (2. 設定したレートでHTTPリクエスト配信) ────┐
                                                                       ▼
                                                              [ Cloud Run (Worker) ]
                                                                       │ (3. 外部API呼び出し)
                                                                       ▼
                                                                  [ 外部API ]

gcloud services enable cloudtasks.googleapis.com

# キューの作成
gcloud tasks queues create api-rate-limit-queue \
    --location=asia-northeast1

# 流量制御（レートリミット）の設定
# 例：1秒間に最大2タスク、同時実行数上限2
gcloud tasks queues update api-rate-limit-queue \
    --location=asia-northeast1 \
    --max-dispatches-per-second=2 \
    --max-concurrent-dispatches=2

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
import logging

app = FastAPI()

class TaskPayload(BaseModel):
    user_id: str
    data: str

@app.post("/process-task")
async def process_task(payload: TaskPayload):
    """
    Cloud Tasks から呼び出されるワーカーエンドポイント
    """
    try:
        # ここに外部APIを呼び出す処理を記述
        logging.info(f"外部API呼び出し開始: User {payload.user_id}")
        
        # 擬似的な外部API処理
        # call_external_api(payload.data)
        
        return {"status": "success", "user_id": payload.user_id}
    except Exception as e:
        logging.error(f"処理失敗: {e}")
        # 500系のエラーを返すと Cloud Tasks が設定に基づき自動リトライしてくれる
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="External API call failed"
        )

import json
from google.cloud import tasks_v2

def enqueue_external_api_task(project_id: str, location: str, queue_name: str, worker_url: str, payload: dict):
    """
    Cloud Tasks キューにタスクを追加する
    """
    client = tasks_v2.CloudTasksClient()
    parent = client.queue_path(project_id, location, queue_name)

    # ペイロードの準備
    json_payload = json.dumps(payload).encode()

    task = {
        "http_request": {
            "http_method": tasks_v2.HttpRequest.Method.POST,
            "url": worker_url,
            "headers": {"Content-Type": "application/json"},
            "body": json_payload,
            # Cloud Run 認証用の OIDC トークン設定（セキュアな呼び出し）
            "oidc_token": {
                "service_account_email": f"cloud-tasks-invoker@{project_id}.iam.gserviceaccount.com"
            }
        }
    }

    response = client.create_task(request={"parent": parent, "task": task})
    print(f"Task created: {response.name}")
    return response

gcloud run deploy worker-service \
    --source . \
    --region asia-northeast1 \
    --no-allow-unauthenticated

gcloud iam service-accounts create cloud-tasks-invoker \
    --display-name="Cloud Tasks Invoker Service Account"

gcloud run services add-iam-policy-binding worker-service \
    --region=asia-northeast1 \
    --member="serviceAccount:cloud-tasks-invoker@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/run.invoker"