FastAPI接收带文件与JSON的PUT请求时触发"Stream consumed"错误的解决咨询

Hey there, let's break down why you're hitting that "Stream consumed" error and how to fix it properly!

问题根源 #

When you send a request with both files and JSON using requests.put(..., files=files, json=json), here's what's actually happening:

1. Requests doesn't send a separate application/json body. Instead, it packages everything into a multipart/form-data request—your JSON gets embedded as an additional form field named json.
2. In your current endpoint, after FastAPI parses the UploadFile parameter, the request stream is already fully consumed. Trying to call await request.json() afterwards throws the error because you can't read the same stream twice.

解决方案 #

Instead of trying to read the raw request body for JSON, modify your endpoint to accept the JSON data as a dedicated form field. Here's the updated code:

First, add the necessary imports:

from fastapi import Form
import json

Then update your endpoint function:

@app.put(f"{settings.API_PREFIX}/project")
async def set_project(
    request: Request, 
    file: Optional[UploadFile] = File(None),
    json_data: str = Form(...)  # Accept JSON as a form field
):
    authorization: str = request.headers.get('Authorization')
    
    # Parse the JSON string into a dictionary
    request_json = json.loads(json_data)
    
    # Now you can work with both request_json and file normally
    # Example: print(request_json["uuid"], file.filename if file else "No file uploaded")
    
    return {"status": "success", "received_uuid": request_json["uuid"], "file_received": file is not None}

为什么这样能解决问题？ #

By accepting the JSON as a form field, FastAPI parses both the file and JSON data from the single multipart/form-data stream in one pass. This avoids trying to re-read the already consumed stream, eliminating the "Stream consumed" error entirely.

测试代码说明 #

Your existing test code doesn't need any changes! When you use requests.put(..., files=files, json=json), Requests automatically encodes the JSON into the json_data form field that our updated endpoint expects.

If you ever want to be explicit, you could rewrite the test to include the JSON in the data parameter instead:

data = {'json_data': json.dumps(json_payload)}
resp = requests.put(url=self.url, headers={"Authorization": TOKEN}, files=files, data=data)

But the original test code works perfectly fine as-is.

备注：内容来源于stack exchange，提问作者Марина Лисниченко