Cloud Storage上の大量のオブジェクトを並行処理でコピーしてみた

from google.cloud import storage
from flask import jsonify, Request
import functions_framework
import logging
from concurrent.futures import ThreadPoolExecutor
import time

def list_files_in_bucket(bucket_name):
    """
    指定されたバケット内のすべてのファイルをリスト取得
    """
    try:
        client = storage.Client()
        bucket = client.bucket(bucket_name)
        blobs = bucket.list_blobs()
        return [blob.name for blob in blobs]
    except Exception as e:
        logging.error(f"Failed to list files in bucket {bucket_name}: {e}")
        raise

def copy_blob(source_bucket_name, destination_bucket_name, source_blob_name, destination_blob_name):
    """
    バケット間でオブジェクトをコピーします。
    """
    try:
        client = storage.Client()
        source_bucket = client.bucket(source_bucket_name)
        destination_bucket = client.bucket(destination_bucket_name)
        source_blob = source_bucket.blob(source_blob_name)

        # オブジェクトをコピー
        source_bucket.copy_blob(source_blob, destination_bucket, destination_blob_name)
        logging.info(f"Copied {source_blob_name} to {destination_blob_name} in bucket {destination_bucket_name}")
    except Exception as e:
        logging.error(f"Failed to copy {source_blob_name} to {destination_blob_name}: {e}")
        raise

def copy_files_in_parallel(source_bucket_name, destination_bucket_name, files, max_workers=1):
    """
    ファイルを並行してコピー
    """
    success_count = 0
    failed_files = []

    def safe_copy(file_name):
        nonlocal success_count
        try:
            destination_blob_name = file_name
            copy_blob(source_bucket_name, destination_bucket_name, file_name, destination_blob_name)
            success_count += 1
        except Exception:
            logging.error(f"Error occurred while processing file: {file_name}")
            failed_files.append(file_name)

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 並行してファイルをコピー
        executor.map(safe_copy, files)
    return success_count, failed_files

@functions_framework.http
def copy_and_batch_delete_all_files(request: Request):
    try:
        request_json = request.get_json()
        if not request_json:
            return jsonify({"error": "Invalid request, JSON body is required."}), 400

        source_bucket = request_json.get("source_bucket")
        destination_bucket = request_json.get("destination_bucket")
        max_workers = request_json.get("max_workers", 10)

        if not source_bucket or not destination_bucket:
            return jsonify({"error": "Missing required parameters: source_bucket, destination_bucket"}), 400

        files = list_files_in_bucket(source_bucket)
        if not files:
            return jsonify({"message": f"No files found in bucket {source_bucket}."}), 200

        # コピー処理の時間を計測
        start_copy_time = time.time()
        success_count, failed_files = copy_files_in_parallel(source_bucket, destination_bucket, files, max_workers)
        end_copy_time = time.time()
        copy_duration = end_copy_time - start_copy_time

        # レスポンスメッセージを変更
        if failed_files:
            return jsonify({
                "message": "Some files failed to copy",
                "copy_duration_seconds": copy_duration,
                "success_count": success_count,
                "failed_count": len(failed_files)
            }), 200
        else:
            return jsonify({
                "message": "All files copied successfully",
                "copy_duration_seconds": copy_duration,
                "success_count": success_count,
                "failed_count": 0
            }), 200

    except Exception as e:
        logging.error(f"Error in Cloud Function: {e}")
        return jsonify({"error": str(e)}), 500```

def copy_files_in_parallel(source_bucket_name, destination_bucket_name, files, max_workers=10):
    """
    ファイルを並行してコピー
    """
    success_count = 0
    failed_files = []

    def safe_copy(file_name):
        nonlocal success_count
        try:
            destination_blob_name = file_name
            copy_blob(source_bucket_name, destination_bucket_name, file_name, destination_blob_name)
            success_count += 1
        except Exception:
            logging.error(f"Error occurred while processing file: {file_name}")
            failed_files.append(file_name)

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 並行してファイルをコピー
        executor.map(safe_copy, files)
    return success_count, failed_files

curl -X POST Cloud FunctionsのURL \
-H "Authorization: bearer $(gcloud auth print-identity-token)" \
-H "Content-Type: application/json" \
-d '{"source_bucket": "コピー元バケット名",  "destination_bucket": "コピー先バケット名",  "max_workers": ワーカー数 }'