Cloud Schedulerで定時にジョブを実行する

import os
import datetime
from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/")
async def run_batch_job(request: Request):
    """
    Cloud Tasks からのPOSTリクエストを受け取り、バッチジョブを実行します。
    """
    now = datetime.datetime.now()
    print(f"[{now}] Batch job started!")
    
    # リクエストボディをJSONとして読み込む (Cloud Tasks からのペイロード)
    try:
        request_body = await request.json()
        print(f"Received payload: {request_body}")
    except Exception as e:
        print(f"Could not parse request body as JSON: {e}")
        request_body = {} # JSONでない場合は空の辞書とする
    
    # ここに実際のバッチ処理ロジックを記述します。
    # 例: データベースの更新、データ処理、外部APIコールなど
    print(f"[{now}] Executing actual batch logic with message: {request_body.get('message', 'No message')}")
    
    print(f"[{now}] Batch job finished!")
    return {"message": "Batch job executed successfully", "timestamp": now.isoformat()}

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

FROM python:3.12-slim

COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv

COPY . /app

WORKDIR /app
RUN uv sync --frozen --no-cache

# Run the application.
CMD ["/app/.venv/bin/fastapi", "run", "app/main.py", "--port", "8080", "--host", "0.0.0.0"]

uv add fastapi[standard]

gcloud auth configure-docker asia-northeast1-docker.pkg.dev
docker build -t asia-northeast1-docker.pkg.dev/[PROJECT_ID]/gcp-sched-tasks-run-sample/image1:0.0.1 .
docker push asia-northeast1-docker.pkg.dev/[PROJECT_ID]/gcp-sched-tasks-run-sample/image1:0.0.1

gcloud run deploy gcp-sched-tasks-run-sample \
--image asia-northeast1-docker.pkg.dev/[PROJECT_ID]/gcp-sched-tasks-run-sample/image1:0.0.1 \
--platform managed \
--region asia-northeast1 \
--allow-unauthenticated # Cloud Tasks からのアクセスを許可するため

gcloud tasks queues create gcp-sched-tasks-run-sample \
--location asia-northeast1

import os
import json
import datetime
from google.cloud import tasks_v2
from google.protobuf import timestamp_pb2

# 環境変数からプロジェクトID、ロケーション、キュー名、Cloud Run URLを取得
project = os.environ.get('GCP_PROJECT')
location = os.environ.get('CLOUD_TASKS_LOCATION', 'asia-northeast1') # Cloud Tasks キューの場所
queue = os.environ.get('CLOUD_TASKS_QUEUE_NAME', 'gcp-sched-tasks-run-sample') # 作成したCloud Tasks キューの名前
cloud_run_url = os.environ.get('CLOUD_RUN_SERVICE_URL') # Cloud Run サービスのURL

client = tasks_v2.CloudTasksClient()
parent = client.queue_path(project, location, queue)

def add_task_to_cloud_tasks(request):
    """Cloud Scheduler からのHTTPリクエストを受け取り、Cloud Tasks にタスクを追加する"""
    if not cloud_run_url:
        print("Error: CLOUD_RUN_SERVICE_URL environment variable is not set.")
        return "Internal Server Error: Cloud Run URL not configured", 500

    request_json = request.get_json(silent=True)
    
    # Cloud Run へ渡したいペイロードがあればここで設定
    # Cloud Scheduler からの任意のデータを Cloud Run へ渡すことができます
    payload_data = {"message": "Hello from Cloud Scheduler (via Cloud Functions)!", "triggered_at": datetime.datetime.now().isoformat()}
    
    # タスクの作成
    task = {
        'http_request': {
            'http_method': tasks_v2.HttpMethod.POST,
            'url': cloud_run_url,
            'headers': {
                'Content-Type': 'application/json'
            },
            'body': json.dumps(payload_data).encode('utf-8')
        }
    }

    # タスクをキューに追加
    try:
        response = client.create_task(parent=parent, task=task)
        print(f"Task created: {response.name}")
        return "Task added to Cloud Tasks successfully", 200
    except Exception as e:
        print(f"Error creating task: {e}")
        return f"Error adding task to Cloud Tasks: {e}", 500

google-cloud-tasks==2.15.0 # Cloud Tasks APIと連携するためのライブラリ
google-cloud-storage # もしCloud Storageを使うなら (今回は必須ではないですが例として) 
Flask # request.get_json() を使う場合は必要 (ただし、Cloud Functionsのトリガーの場合は自動的に提供されることが多い)

gcloud functions deploy cloud_tasks_http_endpoint_fastapi \
  --runtime python313 \
  --trigger-http \
  --allow-unauthenticated \
  --entry-point add_task_to_cloud_tasks \
  --region asia-northeast1 \
  --set-env-vars GCP_PROJECT=[PROJECT_ID],CLOUD_TASKS_LOCATION=asia-northeast1,CLOUD_TASKS_QUEUE_NAME=gcp-sched-tasks-run-sample,CLOUD_RUN_SERVICE_URL=[YOUR_CLOUD_RUN_SERVICE_URL]

gcloud scheduler jobs create http gcp-sched-tasks-run-sample-job \
  --schedule "0 0 * * *" \
  --uri "https://[REGION]-[PROJECT_ID].cloudfunctions.net/cloud_tasks_http_endpoint_fastapi" \
  --http-method POST \
  --message-body "{}" \
  --headers "Content-Type=application/json" \
  --oauth-service-account-email [CLOUD_SCHEDULER_SERVICE_ACCOUNT_EMAIL] \
  --location asia-northeast1