如何使用FastAPI的request.stream()将文件分块上传至Google Cloud Storage？

我完全懂你现在的困境——FastAPI的request.stream()返回的是异步生成器，但gcloud.aio.storage的上传接口根本不认这个类型，它只接受字节、字符串或者基于io.IOBase的类文件对象，而且因为是异步客户端，这个类文件对象还得支持异步读取才行。

咱们可以自己写一个简单的包装类，把FastAPI请求的异步生成器转换成符合要求的异步类文件对象。这样就能实现边读请求体边上传到GCS，完全不用把整个大文件加载到内存里。

首先定义这个包装类：

import io
from typing import AsyncGenerator

class AsyncRequestStream(io.IOBase):
    def __init__(self, stream: AsyncGenerator[bytes, None]):
        self.stream = stream
        self.buffer = b""

    async def read(self, n: int = -1) -> bytes:
        # 先返回缓冲区里剩余的数据
        if self.buffer and n != 0:
            if n == -1 or len(self.buffer) <= n:
                data = self.buffer
                self.buffer = b""
                return data
            else:
                data = self.buffer[:n]
                self.buffer = self.buffer[n:]
                return data
        
        # 从请求的异步生成器里读取下一块数据
        try:
            chunk = await anext(self.stream)
        except StopAsyncIteration:
            return b""
        
        if n == -1:
            return chunk
        
        # 如果读取的块比需要的大，把多余的部分存进缓冲区
        if len(chunk) > n:
            self.buffer = chunk[n:]
            return chunk[:n]
        return chunk

然后修改你的上传函数，把请求流包装成这个类的实例再传给GCS客户端：

from fastapi import Request
import gcloud.aio.storage

async def upload(request: Request, storage_client: gcloud.aio.storage.Storage):
    # 包装request的异步生成器
    upload_stream = AsyncRequestStream(request.stream())
    
    # 最好带上Content-Length，让GCS提前知道文件大小（如果请求里有的话）
    content_length = request.headers.get("Content-Length")
    upload_kwargs = {}
    if content_length:
        upload_kwargs["content_length"] = int(content_length)
    
    await storage_client.upload(
        "my_bucket",
        "my_object",
        upload_stream,
        **upload_kwargs
    )

这个包装类的工作逻辑很简单：它维护了一个小缓冲区，每次调用read()时，先返回缓冲区里剩下的数据；如果缓冲区空了，就从请求的异步生成器里读取下一块数据，按需切割后返回，多余的部分留到下一次读取。这样就完美适配了gcloud.aio.storage对异步类文件对象的要求，全程都是流式处理，内存占用特别低，完全适合处理超大文件上传。

备注：内容来源于stack exchange，提问作者ChrisC