PythonでCloud Tasksを使ってみた

Tasksを使った理由

データ量が膨大すぎたのでCloud Tasksを使って、分割させるため

前提

以前作成した下記記事の環境からスタート

コンソール

Google Cloud PlatformのCloud Tasksからpush キューの作成を行う
キュー名：test
リージョン：asia-northeast1(東京)

コード

requirements.txt

google-cloud-tasks

上記を追加

app.py

import json
import os
from flask import Flask, request
from google.cloud import tasks_v2

app = Flask(__name__)

values = [1,2,3,4,5,6,7,8,9]

# tasksにセット
@app.route('/set_tasks', methods=['POST'])
def set_tasks():
    
    executions_data = []
    # 3つずつ分割する
    for i in range(0, len(values), 3):
        value = {"data":values[i: i + 3]}
        executions_data.append(value)
        
    for data in executions_data:
        client = tasks_v2.CloudTasksClient()
        project = 'プロジェクト名'
        location = 'asia-northeast1'
        queue = 'コンソールで作成したキュー名：test'
        parent = client.queue_path(project, location, queue)
        task = {
            "http_request": {  
                "http_method": tasks_v2.HttpMethod.POST,
                # TODO Runにしたら'RunのURL' + '今後の処理(例：/update)'
                "url": 'https://localhost:0000',  
            }
        }
        task["http_request"]["headers"] = {"Content-type": "application/json"}
        task["http_request"]["body"] = json.dumps(data).encode()
        client.create_task(request={"parent": parent, "task": task})
    return "ok"

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

※ 今回はvaluesに適当な値を入れています

確認

1. ターミナルでpython3 app.pyを実行
2. 別ターミナルでcurl -X POST http://localhost:8080/set_tasksを叩く
   → Cloud Tasksのコンソールで、データ3個のタスクが3つあれば良い！
   今回では3つのタスク、ペイロードに{"data": [1, 2, 3]}らが入っていればOK！

Tasksからデータを受け取る

@app.route("/update", methods=['POST'])
def update():
    tasks_data = request.json["data"]

先ほど記載した'RunのURL' + '今後の処理(例：/update)'で次の処理が上記
routeの/updateから処理が可能
また、request.json["data"]でペイロードにあった{"data": [1, 2, 3]}らを使える

最後に

自分はスプレットシートの大量更新(800行以上)を行なっていたのですが、一括で更新するには時間がかかりすぎたので、Tasksを使用しました。
何か大量の処理をする場合、使えるかもしれません。

参考文献