GCS composeで32以上のオブジェクトをまとめる方法

from concurrent.futures import ALL_COMPLETED, ThreadPoolExecutor, wait

def delete_objects_concurrent(blobs, executor, client) -> None:
    """Delete Cloud Storage objects concurrently.
    Args:
        blobs (List[storage.Blob]): The objects to delete.
        executor (Executor): An executor to schedule the deletions in.
        client (storage.Client): Cloud Storage client to use.
    """
    futures = []
    for blob in blobs:
        logger.info("Deleting slice {}".format(blob.name))
        futures.append(executor.submit(blob.delete, client=client))
    wait(futures, return_when=ALL_COMPLETED)
    logger.info(f"Deleted {len(blobs)} objects")

import logging
from concurrent.futures import ALL_COMPLETED, ThreadPoolExecutor, wait
from itertools import count
from typing import Iterable, List

import google.cloud.logging
from google.cloud import storage

client = google.cloud.logging.Client()
client.setup_logging()

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s %(levelname)s:%(name)s %(message)s"
)
logger = logging.getLogger(__name__)


def generate_composition_chunks(slices: List, chunk_size: int = 31) -> Iterable[List]:
    """Given an indefinitely long list of blobs, return the list in 31-item chunks.

    Arguments:
        slices {List} -- A list of blobs, which are slices of a desired final blob.

    Returns:
        Iterable[List] -- An iteration of 31-item chunks of the input list.

    Yields:
        Iterable[List] -- A 31-item chunk of the input list.
    """
    while len(slices):
        chunk = slices[:chunk_size]
        yield chunk
        slices = slices[chunk_size:]


def generate_hex_sequence() -> Iterable[str]:
    """Generate an indefinite sequence of hexadecimal integers.

    Yields:
        Iterator[Iterable[str]]: The sequence of hex digits, as strings.
    """
    for i in count(0):
        yield hex(i)[2:]

def compose_concurrent(
    object_path: str,
    chunk: List[storage.Blob],
    client: storage.Client,
) -> storage.Blob:
    """

    Arguments:
        object_path {str} -- The path for the composed blob.
        chunk {List[storage.Blob]} -- A list of the blobs that should compose the blob.
        client {storage.Client} -- A Cloud Storage client to use.
    """
    logger.info(f"Composing {object_path}")
    blob = storage.Blob.from_string(object_path)
    blob.compose(chunk, client=client)
    return blob


def compose(
    bucket_name: str,
    prefix: str,
    object_path: str,
    blobs: List[storage.Blob],
    client: storage.Client,
) -> storage.Blob:
    """Compose an object from an indefinite number of blobs. Composition is
    performed concurrently using a tree of accumulators. Do not Cleanup for simplicity.

    Arguments:
        object_path {str} -- The path for the final composed blob.
        buckets {str} -- The bucket name.
        prefix {str} -- The prefix of the blobs.
        blobs {List[storage.Blob]} -- A list of the slices that should
            compose the blob, in order.
        client {storage.Client} -- A Cloud Storage client to use.

    Returns:
        storage.Blob -- The composed blob.
    """
    logger.info("Composing")

    intermediate_prefix = prefix
    identifier = generate_hex_sequence()
    while len(blobs) > 32:
        chunks = generate_composition_chunks(blobs)
        intermediate_prefix = f"{intermediate_prefix}/intermediate"

        with ThreadPoolExecutor(max_workers=4) as executor:
            futures = []
            for chunk in chunks:
                futures.append(
                    executor.submit(
                        compose_concurrent,
                        f"gs://{bucket_name}/{intermediate_prefix}/{next(identifier)}",
                        chunk,
                        client,
                    )
                )
            wait(futures, return_when=ALL_COMPLETED)
        blobs = list(client.list_blobs(bucket_name, prefix=intermediate_prefix))

    # Now can do final compose
    compose_concurrent(object_path, blobs, client)
    logger.info("Composition complete")


def main(request) -> tuple[str, int]:
    """Main entrypoint for the function."""
    project_id = "your_project_id"
    bucket_name = "your_bucket_name"
    prefix = "your_prefix"
    final_blob_name = 'result.csv'
    
    storage_clinet = storage.Client(project_id)
    blobs = list(storage_clinet.list_blobs(bucket_name, prefix=prefix))

    compose(
	object_path=f"gs://{bucket_name}/{prefix}/{final_blob_name}",
	bucket_name=bucket_name,
	prefix=prefix,
	blobs=blobs,
	client=storage_clinet,
    )