背景

通常のCloudRunでは最大実行時間が60分に制限されていますが、それ以上の時間がかかる処理を実行する環境が必要です
コンテナベースで利用時間に応じた従量課金のCloudRunJobsがその解決策として適しています
しかし、CloudRunJobsではHTTPリクエストが許可されておらず、APIのようにPythonなどのアプリからメソッドや引数を指定して処理を呼び出すことができません
そこで今回は、CloudRunJobs実行時に設定できる環境変数を活用して、メソッド名と引数を指定することで処理を呼び出せるようにする方法をご紹介します

CloudRunJobsの実装

import os
import json
from typing import Any


def method_1(args_dict: dict[str, Any]):
    print("method_1")
    print(f"args: {args_dict}")


def method_2(args_dict: dict[str, Any]):
    print("method_2")
    print(f"args: {args_dict}")


def main():
    method = os.getenv('METHOD', None)
    arg_json = os.getenv('ARG_JSON', '{}')

    if method is None:
        raise Exception('env: METHOD has to be given.')

    try:
        arg_dict = json.loads(arg_json)
    except Exception as e:
        print(f'ARG_JSON is not JSON format. : {arg_json}')
        raise e

    if method == 'method_1':
        method_1(arg_dict)
    elif method == 'method_2':
        method_2(arg_dict)
    else:
        raise Exception(f'method_name: {method} not found.')


if __name__ == '__main__':
    main()

METHODという環境変数に対象のメソッド名を実行時に指定します
ARGS_JSONという環境変数に処理の引数となるJSON文字列を指定します

CloudRunJobsの実行

from google.cloud.run_v2 import JobsClient, RunJobRequest, EnvVar

# 環境変数で METHOD と ARG_JSONを渡して呼び出してください
# METHODは必須です
# ARG_JSONはなくてもいいです. 渡す場合は必ずJSONフォーマットで渡してください


def sample_run_job():
    client = JobsClient()

    request = RunJobRequest(
        name=f"projects/{PROJECT}/locations/{REGION}/jobs/{JOB_NAME}",
        overrides=RunJobRequest.Overrides(
            container_overrides=[
                RunJobRequest.Overrides.ContainerOverride(
                    env=[
                        EnvVar(
                            name="METHOD",
                            value="method_1"),
                        EnvVar(
                            name="ARG_JSON",
                            value='{"key1": "value", "key2": ["nest_key", "nest_value"]}')])]))

    operation = client.run_job(request=request)
    print("Waiting for operation to complete...")
    response = operation.result()
    print(response)


sample_run_job()

参考

CloudRunについて

quickstart

docs

PythonからのJobの実行

JobやTaskやExecutionについての理解

Pythonからの実行方法についての解説

その他さまざまな実装の紹介