如何在FastAPI中分片上传文件且不存储到硬盘？

现有代码的内存/硬盘行为说明 #

你当前的代码中：

transferred_data = await request.form()
file_chunk = transferred_data['file_chunk'].file.read()

• await request.form()会触发Starlette的MultiPartParser解析请求。默认情况下，单个文件字段超过1MB时，Starlette会自动将文件内容写入系统临时文件，此时transferred_data['file_chunk'].file指向的是硬盘上的临时文件对象，而非内存中的BytesIO。
• file.read()会把临时文件的内容读取到内存中，但硬盘上的临时文件并不会立即删除，直到UploadFile对象被GC回收后才会自动清理。所以你的代码还是会产生临时文件，不符合“不存储到硬盘”的需求。

让分片完全保留在内存的方案 #

要避免临时文件生成，核心是修改Starlette对大文件的处理逻辑，以下是两种可行方案：

方案1：修改Starlette的MultiPartParser配置 #

Starlette的MultiPartParser有max_file_size参数（默认1MB），超过该值的文件会被写入临时文件。你可以通过自定义Request类调整这个值，将其设置为你的分片大小（比如你用的5MB）：

from fastapi import FastAPI, Request
from starlette.formparsers import MultiPartParser
from starlette.requests import Request as StarletteRequest
from pydantic import parse_obj_as
from pydantic.dataclasses import dataclass
import json

class CustomRequest(StarletteRequest):
    async def form(self):
        if not hasattr(self, "_form"):
            parser = MultiPartParser(
                self.scope,
                self.receive,
                max_file_size=5*1024*1024,  # 设置为你的分片大小
                max_fields=1000,
            )
            self._form = await parser.parse()
        return self._form

app = FastAPI(request_class=CustomRequest)

@dataclass
class Payload:
    part: str
    total_size: int
    chunk_size: int

@app.post('/upload')
async def upload(request: Request):
    filename = request.headers.get('filename')
    transferred_data = await request.form()
    # 此时file_chunk直接来自内存中的BytesIO
    file_chunk = transferred_data['file_chunk'].file.read()
    json_loads = json.loads(transferred_data['payload'])
    payload = parse_obj_as(Payload, json_loads)
    # 这里直接转发file_chunk到云端即可
    return {"status": "success", "part": payload.part}

方案2：手动解析Multipart请求体 #

如果不想依赖Starlette的默认form解析，可以直接读取请求体并手动解析multipart数据，完全控制文件内容的存储：

from fastapi import FastAPI, Request
from pydantic import parse_obj_as
from pydantic.dataclasses import dataclass
import json
import email

@dataclass
class Payload:
    part: str
    total_size: int
    chunk_size: int

app = FastAPI()

@app.post('/upload')
async def upload(request: Request):
    filename = request.headers.get('filename')
    content_type = request.headers.get('Content-Type')
    # 解析multipart边界
    boundary = content_type.split('boundary=')[1]
    # 读取全部请求体到内存
    body = await request.body()
    # 用email模块解析multipart数据
    msg = email.message_from_bytes(body)
    payload_data = None
    file_chunk = None
    for part in msg.walk():
        if part.is_multipart():
            continue
        field_name = part.get_param('name', header='Content-Disposition')
        if field_name == 'payload':
            payload_data = json.loads(part.get_payload(decode=True))
        elif field_name == 'file_chunk':
            file_chunk = part.get_payload(decode=True)
    
    payload = parse_obj_as(Payload, payload_data)
    # 这里直接使用内存中的file_chunk转发到云端
    return {"status": "success", "part": payload.part}

检查FastAPI硬盘使用的方法 #

1. 查看临时文件路径：在现有代码中，打印transferred_data['file_chunk'].file.name即可获取临时文件的绝对路径，直接去对应目录查看是否有文件生成：

print(transferred_data['file_chunk'].file.name)

2. 监控系统临时目录：Starlette默认使用系统临时目录（Linux为/tmp，Windows为%TEMP%），可以通过工具实时监控该目录下的文件变化，比如Linux用watch ls -l /tmp，Windows直接打开临时目录查看。
3. 用代码监控文件句柄：使用psutil库查看FastAPI进程打开的文件，判断是否有临时文件被创建：

import psutil
import os

proc = psutil.Process(os.getpid())
for file in proc.open_files():
    if 'tmp' in file.path:
        print(f"临时文件：{file.path}")

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