Cloud Run FunctionsからAWSのリソース(S3)をアクセスキー払い出しなしで操作する

# サービスアカウント作成コマンド
gcloud iam service-accounts create サービスアカウント名 \
--description="AWS操作用" \
--display-name="AWS Access"

# uniqueId取得コマンド
gcloud iam service-accounts describe サービスアカウント名@プロジェクト名.iam.gserviceaccount.com --format="json" | jq .uniqueId

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Federated": "accounts.google.com"
            },
            "Action": "sts:AssumeRoleWithWebIdentity",
            "Condition": {
                "StringEquals": {
                    "accounts.google.com:oaud": "適当な値",
                    "accounts.google.com:sub": "一意なID"
                }
            }
        }
    ]
}

# S3の指定パス配下のparquetファイルを読み込んでサマリを返すスクリプト
import functions_framework
import boto3
import awswrangler as wr
import google.auth
import google.auth.transport.requests
from google.auth import compute_engine

# AWS IAM ロールの ARN
AWS_ROLE_ARN = "IAMロールのARN"
CLOUD_RUN_URL = "Cloud Run FunctionsのURL"

def get_google_identity_token():
    """メタデータサーバーを使用して ID トークンを取得"""
    request = google.auth.transport.requests.Request()

    credentials = compute_engine.IDTokenCredentials(
        request=request, target_audience=CLOUD_RUN_URL, use_metadata_identity_endpoint=True
    )

    credentials.refresh(request)
    return credentials.token

def get_aws_credentials():
    """Google Cloud の ID トークンを使用して AWS の一時認証情報を取得"""
    identity_token = get_google_identity_token()

    sts_client = boto3.client("sts")
    response = sts_client.assume_role_with_web_identity(
        RoleArn=AWS_ROLE_ARN,
        RoleSessionName="cloud-run-session",
        WebIdentityToken=identity_token
    )

    return response["Credentials"]

def fetch_s3_parquet_data(bucket: str, prefix: str):
    """指定した S3 パスの Parquet ファイルを一括で読み込み、Pandas DataFrame に変換"""
    aws_creds = get_aws_credentials()

    session = boto3.Session(
        aws_access_key_id=aws_creds["AccessKeyId"],
        aws_secret_access_key=aws_creds["SecretAccessKey"],
        aws_session_token=aws_creds["SessionToken"]
    )

    s3_path = f"s3://{bucket}/{prefix}*"
    print(f"Fetching Parquet files from: {s3_path}")

    # Parquet ファイルを一括で読み込む
    df = wr.s3.read_parquet(path=s3_path, boto3_session=session)
    return df

@functions_framework.http
def main(request):
    bucket = "S3バケット名" 
    prefix = "ファイルが保存されているパス"

    try:
        df = fetch_s3_parquet_data(bucket, prefix) 
        return {"rows": df.shape[0], "columns": df.shape[1]}, 200
    except Exception as e:
        return {"error": str(e)}, 500

google-auth
google-auth-oauthlib
google-auth-httplib2
boto3
awswrangler
functions-framework==3.*

def get_google_identity_token():
    """メタデータサーバーを使用して ID トークンを取得"""
    request = google.auth.transport.requests.Request()

    credentials = compute_engine.IDTokenCredentials(
        request=request,  target_audience=CLOUD_RUN_URL, use_metadata_identity_endpoint=True
    )

    credentials.refresh(request)
    return credentials.token

def get_aws_credentials():
    """Google Cloud の ID トークンを使用して AWS の一時認証情報を取得"""
    identity_token = get_google_identity_token()  # ID トークンを取得

    sts_client = boto3.client("sts")
    response = sts_client.assume_role_with_web_identity(
        RoleArn=AWS_ROLE_ARN,
        RoleSessionName="cloud-run-session",
        WebIdentityToken=identity_token  # ID トークンを渡す
    )

    return response["Credentials"]

def fetch_s3_parquet_data(bucket: str, prefix: str):
    """指定した S3 パスの Parquet ファイルを一括で読み込み、Pandas DataFrame に変換"""
    aws_creds = get_aws_credentials()

    session = boto3.Session(
        aws_access_key_id=aws_creds["AccessKeyId"],
        aws_secret_access_key=aws_creds["SecretAccessKey"],
        aws_session_token=aws_creds["SessionToken"]
    )

    s3_path = f"s3://{bucket}/{prefix}*"  # 指定パス内のすべての Parquet ファイルを対象
    print(f"Fetching Parquet files from: {s3_path}")  # デバッグ用ログ

    # Parquet ファイルを一括で読み込む
    df = wr.s3.read_parquet(path=s3_path, boto3_session=session)

    print(f"DataFrame shape: {df.shape}")  # データの行数・列数を確認
    print(df.head())  # 最初の数行を表示

    return df

data = {
    "id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35],
    "score": [85.5, 90.0, 78.5]
}

# 実行コマンド
curl -m 370 -X POST "Cloud Run FunctionsのURL" -H "Authorization: bearer $(gcloud auth print-identity-token)"

{"columns":4,"rows":3}