如何基于FastAPI+MinIO Python SDK流式分片上传大文件

核心思路 #

要实现边接收边上传至MinIO且避免分片覆盖，需使用MinIO的**Multipart Upload（多部分上传）**机制：

1. 先初始化一个多部分上传会话，获取唯一upload_id
2. 将FastAPI流式接收的文件数据按固定大小拆分成分片，为每个分片分配唯一的part_number
3. 每收到一个分片就立即调用MinIO的分片上传接口，上传后记录该分片的ETag
4. 所有分片上传完成后，提交分片列表完成合并，最终得到完整对象

具体实现代码 #

1. 初始化异步MinIO客户端 #

from minio.aio import Minio
from minio.errors import S3Error

minio_client = Minio(
    "your-minio-endpoint:port",
    access_key="your-access-key",
    secret_key="your-secret-key",
    secure=False  # 根据实际部署情况设置是否启用HTTPS
)

2. 修改FastAPI接口实现流式分片上传 #

from fastapi import FastAPI, Request
from streaming_form_data import StreamingFormDataParser
from streaming_form_data.targets import ValueTarget

app = FastAPI()

# 定义分片大小（MinIO要求非最后一个分片至少5MB，此处设为5*1024*1024字节）
CHUNK_SIZE = 5 * 1024 * 1024

@app.post("/upload")
async def stream_upload_to_minio(request: Request):
    # 从请求头或表单字段获取文件名、对象路径、存储桶名（可根据实际业务调整）
    filename = request.headers.get("X-Filename")
    object_path = request.headers.get("X-Object-Path")
    bucket_name = "your-bucket-name"

    try:
        # 1. 初始化MinIO多部分上传会话
        upload_id = await minio_client.create_multipart_upload_async(
            bucket_name, object_path
        )

        parser = StreamingFormDataParser(headers=request.headers)
        # 用ValueTarget接收文件流，跳过本地存储步骤
        file_target = ValueTarget()
        parser.register("file", file_target)

        part_number = 1
        parts = []
        current_chunk = b""

        async for chunk in request.stream():
            # 解析表单数据，提取文件内容
            parser.data_received(chunk)
            # 累积数据到当前分片
            current_chunk += file_target.value
            file_target.value = b""  # 清空target，避免重复累积

            # 当当前分片达到设定大小，立即上传至MinIO
            if len(current_chunk) >= CHUNK_SIZE:
                etag = await minio_client.upload_part_async(
                    bucket_name,
                    object_path,
                    upload_id,
                    part_number,
                    current_chunk
                )
                parts.append({"PartNumber": part_number, "ETag": etag})
                part_number += 1
                current_chunk = b""

        # 处理最后一个不足CHUNK_SIZE的分片
        if current_chunk:
            etag = await minio_client.upload_part_async(
                bucket_name,
                object_path,
                upload_id,
                part_number,
                current_chunk
            )
            parts.append({"PartNumber": part_number, "ETag": etag})

        # 2. 完成多部分上传，合并所有分片
        await minio_client.complete_multipart_upload_async(
            bucket_name, object_path, upload_id, parts
        )

        return {"status": "success", "message": "文件上传完成"}

    except S3Error as e:
        # 上传失败时取消多部分上传，避免MinIO残留无用分片
        await minio_client.abort_multipart_upload_async(
            bucket_name, object_path, upload_id
        )
        return {"status": "error", "message": str(e)}

关键注意事项 #

• 分片大小限制：MinIO默认要求除最后一个分片外，其余分片大小至少为5MB，若需更小分片需修改MinIO服务器配置（不推荐）
• 异步客户端使用：必须使用minio.aio下的异步方法，否则会阻塞FastAPI事件循环，影响接口性能
• 表单解析：需确保StreamingFormDataParser注册的字段名与前端上传的表单字段一致（示例中为"file"）
• 异常处理：上传失败时务必调用abort_multipart_upload_async取消会话，防止MinIO中残留未完成的分片占用存储空间

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