Cloud Tasksでタスク単位のリトライ設定とバッチ作成・削除がGAになったので試してみる

$ python --version
Python 3.13.9
$ gcloud version | head -1
Google Cloud SDK 588.0.0
$ pip list | grep -iE "^(google-auth|requests) "
google-auth        2.60.0
requests           2.34.2

$ pip install google-auth requests
$ gcloud auth application-default login

import os

from flask import Flask, request

app = Flask(__name__)

@app.post("/")
def handler():
    task_name = request.headers.get("X-CloudTasks-TaskName")
    retry_count = request.headers.get("X-CloudTasks-TaskRetryCount")
    print(f"task={task_name} retry_count={retry_count}")
    return "always fail", 500

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))

flask==3.*
gunicorn==23.*

web: gunicorn -b :$PORT main:app

$ gcloud run deploy always-fail --source . --region=asia-northeast1 --no-allow-unauthenticated
$ gcloud iam service-accounts create tasks-invoker
$ gcloud run services add-iam-policy-binding always-fail --region=asia-northeast1 \
    --member="serviceAccount:tasks-invoker@{プロジェクトID}.iam.gserviceaccount.com" \
    --role="roles/run.invoker"

$ gcloud tasks queues create test-retry-queue --location=asia-northeast1 \
    --max-attempts=5 --min-backoff=10s --max-backoff=300s --max-doublings=3 \
    --log-sampling-ratio=1.0

import google.auth
import google.auth.transport.requests

PROJECT_ID = "{プロジェクトID}"
LOCATION = "asia-northeast1"
QUEUE_ID = "test-retry-queue"
TARGET_URL = "https://always-fail-{プロジェクト番号}.asia-northeast1.run.app/"
INVOKER_SA = f"tasks-invoker@{PROJECT_ID}.iam.gserviceaccount.com"

QUEUE_PATH = f"projects/{PROJECT_ID}/locations/{LOCATION}/queues/{QUEUE_ID}"
BASE_URL = f"https://cloudtasks.googleapis.com/v2/{QUEUE_PATH}"

def auth_headers() -> dict:
    credentials, _ = google.auth.default(
        scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )
    credentials.refresh(google.auth.transport.requests.Request())
    return {
        "Authorization": f"Bearer {credentials.token}",
        "Content-Type": "application/json; charset=utf-8",
    }

def http_task(task_id: str, schedule_time: str | None = None) -> dict:
    task = {
        "name": f"{QUEUE_PATH}/tasks/{task_id}",
        "httpRequest": {
            "url": TARGET_URL,
            "httpMethod": "POST",
            "oidcToken": {"serviceAccountEmail": INVOKER_SA},
        },
    }
    if schedule_time:
        task["scheduleTime"] = schedule_time
    return task

import requests

from common import BASE_URL, auth_headers, http_task

headers = auth_headers()

# キューの retryConfig をそのまま使うタスク
default_task = http_task("retry-default")

# タスク単位で retryConfig を上書きするタスク
override_task = http_task("retry-override")
override_task["retryConfig"] = {
    "maxAttempts": 2,
    "minBackoff": "1s",
    "maxBackoff": "1s",
}

for task in [default_task, override_task]:
    res = requests.post(f"{BASE_URL}/tasks", headers=headers, json={"task": task})
    res.raise_for_status()
    print(res.json())

'retryConfig': {'maxAttempts': 2, 'minBackoff': '1s', 'maxBackoff': '1s', 'maxDoublings': 16}

resource.type="cloud_tasks_queue"
resource.labels.queue_id="test-retry-queue"
severity>=DEFAULT

All tasks in a batch request must target the same queue. Found task for queue  in batch request for queue projects/{プロジェクトID}/locations/asia-northeast1/queues/test-retry-queue.

import json
import time
import uuid
from datetime import datetime, timedelta, timezone

import requests

from common import BASE_URL, QUEUE_PATH, auth_headers, http_task

headers = auth_headers()
schedule_time = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()

single_ids = [uuid.uuid4().hex for _ in range(100)]
batch_ids = [uuid.uuid4().hex for _ in range(100)]

# 1件ずつ tasks.create を呼ぶ
start = time.perf_counter()
for task_id in single_ids:
    task = http_task(task_id, schedule_time)
    res = requests.post(f"{BASE_URL}/tasks", headers=headers, json={"task": task})
    res.raise_for_status()
print(f"single create x100: {time.perf_counter() - start:.2f}s")

# tasks:batchCreate で100件をまとめて登録する
body = {
    "requests": [
        {"parent": QUEUE_PATH, "task": http_task(task_id, schedule_time)}
        for task_id in batch_ids
    ]
}
start = time.perf_counter()
res = requests.post(f"{BASE_URL}/tasks:batchCreate", headers=headers, json=body)
res.raise_for_status()
operation = res.json()
print(f"batchCreate request: {time.perf_counter() - start:.2f}s")
print(operation)

# オペレーションの完了を待つ
op_url = f"https://cloudtasks.googleapis.com/v2/{operation['name']}"
while not operation.get("done"):
    time.sleep(1)
    operation = requests.get(op_url, headers=headers).json()
print(f"batchCreate done: {time.perf_counter() - start:.2f}s")
print(operation)

# バッチ削除で使うタスク名を書き出す
with open("created_tasks.json", "w") as f:
    json.dump([f"{QUEUE_PATH}/tasks/{i}" for i in single_ids + batch_ids], f)

$ python batch_create.py
single create x100: 27.45s
batchCreate request: 0.46s
{'name': 'projects/{プロジェクトID}/locations/asia-northeast1/operations/17c34694-094f-46a4-950a-262d935a112a', 'metadata': {'@type': 'type.googleapis.com/google.cloud.tasks.v2.BatchCreateTasksMetadata', ..., 'state': 'SUCCEEDED'}, 'done': True, 'response': {'@type': 'type.googleapis.com/google.cloud.tasks.v2.BatchCreateTasksResponse', 'tasks': [...]}}
batchCreate done: 0.46s
（以下省略）

import uuid
from datetime import datetime, timedelta, timezone

import requests

from common import BASE_URL, QUEUE_PATH, auth_headers, http_task

headers = auth_headers()
schedule_time = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()

# 101件を1回で送る
body = {
    "requests": [
        {"parent": QUEUE_PATH, "task": http_task(uuid.uuid4().hex, schedule_time)}
        for _ in range(101)
    ]
}
res = requests.post(f"{BASE_URL}/tasks:batchCreate", headers=headers, json=body)
print(res.status_code, res.json())

# 先に1件作成しておき、同じ名前と新しい名前を混ぜて送る
dup_id = uuid.uuid4().hex
new_id = uuid.uuid4().hex
res = requests.post(
    f"{BASE_URL}/tasks", headers=headers, json={"task": http_task(dup_id, schedule_time)}
)
res.raise_for_status()
print(f"existing={dup_id} new={new_id}")

body = {
    "requests": [
        {"parent": QUEUE_PATH, "task": http_task(dup_id, schedule_time)},
        {"parent": QUEUE_PATH, "task": http_task(new_id, schedule_time)},
    ]
}
res = requests.post(f"{BASE_URL}/tasks:batchCreate", headers=headers, json=body)
print(res.status_code, res.json())

$ python batch_create_errors.py
400 {'error': {'code': 400, 'message': 'A maximum of 100 tasks can be created in a single batch, but got 101 tasks.', 'status': 'INVALID_ARGUMENT'}}
existing=949eb0dc70bd492eb77dcd0483e93126 new=51a9f38682364f289deb8194a8976619
409 {'error': {'code': 409, 'message': 'Requested entity already exists', 'status': 'ALREADY_EXISTS'}}

import json
import time

import requests

from common import BASE_URL, auth_headers

headers = auth_headers()

with open("created_tasks.json") as f:
    names = json.load(f)

res = requests.post(f"{BASE_URL}/tasks:batchDelete", headers=headers, json={"names": names})
res.raise_for_status()
operation = res.json()

op_url = f"https://cloudtasks.googleapis.com/v2/{operation['name']}"
while not operation.get("done"):
    time.sleep(1)
    operation = requests.get(op_url, headers=headers).json()
print(operation)

$ python batch_delete.py
{'name': 'projects/{プロジェクトID}/locations/asia-northeast1/operations/9ab70545-ddfa-45e5-8b84-1d8978e402c8', 'metadata': {'@type': 'type.googleapis.com/google.cloud.tasks.v2.BatchDeleteTasksMetadata', 'startTime': '2026-10-10T19:15:39.233324648Z', 'endTime': '2026-10-10T19:15:39.291555170Z', 'state': 'SUCCEEDED'}, 'done': True, 'response': {'@type': 'type.googleapis.com/google.protobuf.Empty'}}
$ gcloud tasks list --queue=test-retry-queue --location=asia-northeast1
TASK_NAME                         TYPE  CREATE_TIME           SCHEDULE_TIME                DISPATCH_ATTEMPTS  RESPONSE_ATTEMPTS  LAST_ATTEMPT_STATUS
51a9f38682364f289deb8194a8976619  http  2026-10-10T19:15:08Z  2026-10-10T20:15:07.229857Z  0                  0                  Unknown
949eb0dc70bd492eb77dcd0483e93126  http  2026-10-10T19:15:07Z  2026-10-10T20:15:07.229857Z  0                  0                  Unknown
retry-default                     http  2026-10-10T19:13:08Z  2026-10-10T19:15:41.332400Z  4                  0                  INTERNAL(13): HTTP status code 500

200 {..., 'metadata': {..., 'state': 'PARTIALLY_SUCCEEDED', 'failedRequests': {'0': {'code': 5, 'message': 'Requested entity was not found.'}}}, 'done': True, ...}