如何使用FastAPI下载大文件？
我正在尝试从.tar.gz后端下载一个大文件( FastAPI )。在服务器端，我只是验证文件路径，然后使用Starlette.FileResponse返回整个文件-就像我在StackOverflow上的许多相关问题中看到的一样。
服务器端：
return FileResponse(path=file_name, media_type='application/octet-stream', filename=file_name)在此之后，我得到以下错误：
File "/usr/local/lib/python3.10/dist-packages/fastapi/routing.py", line 149, in serialize_response
return jsonable_encoder(response_content)
File "/usr/local/lib/python3.10/dist-packages/fastapi/encoders.py", line 130, in jsonable_encoder
return ENCODERS_BY_TYPE[type(obj)](obj)
File "pydantic/json.py", line 52, in pydantic.json.lambda
UnicodeDecodeError: 'utf-8' codec can't decode byte 0x8b in position 1: invalid start byte我也尝试使用StreamingResponse，但得到了同样的错误。还有其他方法吗？
代码中的StreamingResponse：
@x.post("/download")
async def download(file_name=Body(), token: str | None = Header(default=None)):
file_name = file_name["file_name"]
# should be something like xx.tar
def iterfile():
with open(file_name,"rb") as f:
yield from f
return StreamingResponse(iterfile(),media_type='application/octet-stream')好的，这是对这个问题的更新。我发现这个api没有出现错误，但是api对此进行了前向请求。
@("/")
def f():
req = requests.post(url ="/download")
return req.content在这里，如果我返回一个带有StreamingResponse文件的.tar，就会导致(可能)编码问题。
使用请求时，请记住设置相同的媒体类型。这是media_type='application/octet-stream'。而且起作用了！
Stack Overflow用户
发布于 2022-09-25 08:59:23
如果发现yield from f在使用类似文件的对象时速度相当慢，则可以创建一个生成器，在该生成器中使用指定的块大小以块形式读取文件；因此，可以加快处理速度。例子见下文。
请注意，StreamingResponse可以使用async生成器或普通生成器/迭代器来流响应体。如果您使用的标准open()方法不支持async/await，则必须使用普通的def声明生成器函数。无论如何，FastAPI/Starlette仍将异步工作，因为它将检查您传递的生成器是否是异步的(如源代码中所示)，如果不是，它将使用iterate_in_threadpool在一个单独的线程中运行生成器，然后等待该线程。
您可以在响应中设置Content-Disposition头(如这个答案以及这里和这里中所描述的)，以指示是否希望在浏览器中显示inline (如果您是流的，例如，.mp4视频、.mp3音频文件等)，或者是本地下载和保存的attachment (使用指定的filename)。
至于media_type (也称为MIME类型)，有两种主要MIME类型(参见常见MIME类型)：
text/plain是文本文件的默认值。文本文件应该是人类可读的，不能包含二进制数据.application/octet-stream是所有其他情况下的默认值。未知文件类型应使用此类型的。
对于一个扩展名为.tar的文件，如您的问题中所示，您还可以使用与octet-stream不同的子类型，即x-tar。否则，如果文件类型未知，请坚持使用application/octet-stream。有关常见MIME类型的列表，请参阅上面的链接文档。
选项1-使用普通生成器
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
CHUNK_SIZE = 1024 * 1024 # = 1MB - adjust the chunk size as desired
some_file_path = 'large_file.tar'
app = FastAPI()
@app.get('/')
def main():
def iterfile():
with open(some_file_path, 'rb') as f:
while chunk := f.read(CHUNK_SIZE):
yield chunk
headers = {'Content-Disposition': 'attachment; filename="large_file.tar"'}
return StreamingResponse(iterfile(), headers=headers, media_type='application/x-tar')选项2-将async生成器与aiofiles一起使用
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import aiofiles
CHUNK_SIZE = 1024 * 1024 # = 1MB - adjust the chunk size as desired
some_file_path = 'large_file.tar'
app = FastAPI()
@app.get('/')
async def main():
async def iterfile():
async with aiofiles.open(some_file_path, 'rb') as f:
while chunk := await f.read(CHUNK_SIZE):
yield chunk
headers = {'Content-Disposition': 'attachment; filename="large_file.tar"'}
return StreamingResponse(iterfile(), headers=headers, media_type='application/x-tar')https://stackoverflow.com/questions/73550398
复制相似问题
腾讯云开发者
