如何在非异步方法中读取异步流？FastAPI PUT请求上传文件遇兼容问题

嘿，这个问题我太熟了！你碰到的本质是同步代码和异步流的API不兼容——FastAPI基于Starlette，在同步路径函数里调用request.stream()返回的是异步生成器，但你原来的WSGI代码只认带read()方法的同步类文件对象，自然会报那个错。

我给你几个实用的解决思路，从简单到进阶都有，不用大改你原来的代码：

方案1：一次性读取全部请求体（适合小文件） #

这个方案最直接，先把异步流的所有内容读出来，转成Python标准库的BytesIO对象（它完全支持read()方法，完美匹配你旧代码的预期）。

首先要用到asgiref的async_to_sync工具（FastAPI默认已经依赖这个库了，不用额外装），它能帮我们在同步函数里安全执行异步操作：

from asgiref.sync import async_to_sync
from io import BytesIO
from fastapi import Request

@f_app.put("/share/{share_name}/file/upload")
def upload_file(request: Request, share_name: str):
    path = request.headers.get("x-file-path", "")
    
    # 用async_to_sync包装异步的body读取，拿到完整的请求体字节
    body_content = async_to_sync(request.body)()
    # 转成支持read()的类文件对象
    sync_file_like = BytesIO(body_content)
    
    # 替换原来的stream传入逻辑
    request.state.api_app.router.input.with_stream(sync_file_like)
    
    return UploadAPIEndpoint(request.state.api_app).action(
        "put", share_name=share_name, path=path
    )

原理很简单：async_to_sync帮我们搞定了同步/异步上下文的切换，BytesIO则模拟了WSGI时代请求体的类文件接口，旧代码完全感知不到变化。

方案2：流式读取（适合大文件，避免内存溢出） #

如果你的上传文件很大，一次性读入内存会有压力，那可以自己写一个包装类，把异步流转换成支持read()方法的同步流，逐块读取内容：

from asgiref.sync import async_to_sync
from io import IOBase
from fastapi import Request

class AsyncStreamWrapper(IOBase):
    def __init__(self, async_stream):
        self.async_stream = async_stream
        # 把异步迭代器转成同步迭代器
        self.iterator = async_to_sync(lambda: async_stream.__aiter__())()
        self.buffer = b""
    
    def read(self, size=-1):
        if size == -1:
            # 读取全部内容
            content = b""
            for chunk in self.iterator:
                content += chunk
            return content
        # 读取指定大小的内容
        while len(self.buffer) < size:
            try:
                chunk = next(self.iterator)
                self.buffer += chunk
            except StopIteration:
                break  # 流读完了
        # 取出需要的部分，剩下的留到下次读取
        result = self.buffer[:size]
        self.buffer = self.buffer[size:]
        return result

# 然后修改你的路径函数
@f_app.put("/share/{share_name}/file/upload")
def upload_file(request: Request, share_name: str):
    path = request.headers.get("x-file-path", "")
    
    # 把异步流包装成支持read()的同步对象
    sync_stream = AsyncStreamWrapper(request.stream())
    request.state.api_app.router.input.with_stream(sync_stream)
    
    return UploadAPIEndpoint(request.state.api_app).action(
        "put", share_name=share_name, path=path
    )

这个包装类会逐块读取异步流的内容，不会一次性把整个文件加载到内存里，和你原来WSGI的流式处理逻辑一致。

为什么原来的代码会报错？ #

再给你补个小科普：WSGI的请求体是同步的类文件对象，天然支持read()；但FastAPI用的是ASGI标准，在同步路径函数里，request.stream()返回的是异步生成器（只能用async for遍历），它没有read()方法，所以你原来的代码直接传进去就炸了。我们做的就是给异步流套个“壳”，让它看起来和WSGI的请求体一样。

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