FastAPI 流式分段上传

示例说明

UploadFile.read(size) 是流式读取的关键：每次只取一块，await 让出事件循环，并发上传互不阻塞。体积上限必须在写入过程中检查——收到超限时已经落盘的部分要删除临时文件再报错，半成品不能留在存储里。MIME 白名单在最前面拦截不支持的类型。

技术要点

• 分块读取让事件循环保持响应
• 写入过程中即时执行体积上限检查
• 超限路径清理临时文件不留残余

代码解读

上传入口先验类型，再以 512KB 为块循环读写：每块累计字节数，超过 20MB 即关流、删临时文件、返回 413。正常完成后去掉 .part 后缀原子收尾，返回文件名与字节数。aiofiles 让磁盘写入也走异步路径。

使用建议

高并发上传要限制并发数保护磁盘带宽。临时文件目录与最终存储隔离便于清理孤儿。安全场景追加文件魔数校验。

代码示例

# stream multipart parts to disk without buffering the whole file
from pathlib import Path

import aiofiles
from fastapi import FastAPI, File, HTTPException, UploadFile

MAX_BYTES = 20 * 1024 * 1024
CHUNK = 1024 * 512
ALLOWED = {'application/pdf', 'image/png'}

app = FastAPI()
STORAGE = Path('uploads')
STORAGE.mkdir(exist_ok=True)


@app.post('/documents')
async def upload_document(file: UploadFile = File(...)) -> dict:
    if file.content_type not in ALLOWED:
        raise HTTPException(415, f'unsupported type: {file.content_type}')

    written = 0
    target = STORAGE / f'{file.filename or "doc"}.part'

    async with aiofiles.open(target, 'wb') as out:
        while chunk := await file.read(CHUNK):
            written += len(chunk)
            if written > MAX_BYTES:
                await out.close()
                target.unlink(missing_ok=True)
                raise HTTPException(413, 'file exceeds size limit')
            await out.write(chunk)

    final = target.with_suffix('')
    target.rename(final)
    return {'filename': final.name, 'bytes': written, 'type': file.content_type}