FastAPI中间件读取表单后后续路由无法处理数据的解决方案咨询

FastAPI基于Starlette实现，Request对象的请求体是流式结构，调用await request.form()会一次性消费掉请求体的所有内容。后续路由或中间件再尝试读取请求体时，会因为流已耗尽而无法获取数据。直接给request._form赋值无效，因为内部的请求体流已经空了，Starlette依然会尝试从流中读取数据而非直接使用_form。

方案一：复制请求体并复用解析后的表单 #

通过复制原始请求体内容，重新构建可读取的流，同时将解析后的表单存入request._form，让后续流程既能读取请求体，也能直接复用已解析的结果：

from fastapi import FastAPI, Request, UploadFile, Form, File
from starlette.datastructures import UploadFile as StarletteUploadFile
import io

app = FastAPI()

@app.middleware("http")
async def middleware_log(request: Request, call_next):
    content_type = request.headers.get("Content-Type")
    if content_type and "multipart/form-data" in content_type:
        # 复制原始请求体内容
        raw_body = await request.body()
        # 重新设置请求体的字节流，让后续可重复读取
        request._body = raw_body
        request._stream = io.BytesIO(raw_body)
        
        # 解析表单并生成日志数据
        form = await request.form()
        log_data = {
            "files": [
                {
                    "filename": file.filename, 
                    "file_size": str(round(file.size / 1024, 2)) + 'KB', 
                    "content_type": file.content_type
                } for file in form.values() if isinstance(file, StarletteUploadFile)
            ],
            "form_fields": {k: v for k, v in form.items() if not isinstance(v, StarletteUploadFile)}
        }
        print(log_data)
        
        # 将解析后的表单存入request._form，避免后续重复解析
        request._form = form
    
    response = await call_next(request)
    return response

方案二：使用依赖项处理日志（更简洁） #

利用FastAPI的依赖项机制，在路由处理前自动解析表单并记录日志，FastAPI内部会自动缓存解析后的表单，路由中的File/Form参数可直接获取数据：

from fastapi import FastAPI, Request, UploadFile, Form, File, Depends
from starlette.datastructures import UploadFile as StarletteUploadFile

app = FastAPI()

async def log_form_data(request: Request):
    content_type = request.headers.get("Content-Type")
    if content_type and "multipart/form-data" in content_type:
        form = await request.form()
        log_data = {
            "files": [
                {
                    "filename": file.filename, 
                    "file_size": str(round(file.size / 1024, 2)) + 'KB', 
                    "content_type": file.content_type
                } for file in form.values() if isinstance(file, StarletteUploadFile)
            ],
            "form_fields": {k: v for k, v in form.items() if not isinstance(v, StarletteUploadFile)}
        }
        print(log_data)

# 全局添加依赖，所有路由都会执行日志逻辑
app.dependency_overrides[None] = [log_form_data]

# 测试路由示例
@app.post("/upload")
async def upload_file(file: UploadFile = File(...), username: str = Form(...)):
    return {"uploaded_file": file.filename, "username": username}

注意事项 #

• 若处理超大文件，方案一中的内存复制可能导致性能问题，此时可考虑将请求体写入临时文件而非内存。
• 方案二更符合FastAPI的设计规范，无需手动操作请求体的底层属性，代码更简洁易维护。

内容的提问来源于stack exchange，提问作者hbx