使用Python和FastAPI实现MinIO分块上传与断点续传

环境与依赖准备

第1步：安装必要的 Python 库

在虚拟环境中安装 minio 用于对象存储交互，fastapi 与 uvicorn 提供异步 Web 服务。

pip install minio fastapi uvicorn

核心方法封装：分块上传逻辑

第2步：配置 MinIO 客户端与分块上传函数

使用 Minio 类实例化连接参数。为支持大文件传输并预留断点续传基础，通过 initiate_multipart_upload 获取任务 ID，循环读取固定大小的数据块并逐块提交。若途中发生网络或磁盘异常，直接调用 abort_multipart_upload 清理临时分块。

from minio import Minio
from minio.error import S3Error

def minio_client():
    return Minio(
        "play.min.io",
        access_key="your-access-key",
        secret_key="your-secret-key",
        secure=True
    )

def upload_file_with_resume(client, bucket_name, object_name, file_path, part_size=10*1024*1024):
    upload_id = client.initiate_multipart_upload(bucket_name, object_name)
    try:
        with open(file_path, "rb") as file_data:
            part_number = 1
            while True:
                data = file_data.read(part_size)
                if not data:
                    break
                # 注：实际生产环境的完整断点续传需在此处增加已上传分片校验与偏移量计算逻辑
                client.put_object(bucket_name, f"{object_name}.{part_number}", data, len(data), part_number=part_number, upload_id=upload_id)
                part_number += 1
        client.complete_multipart_upload(bucket_name, object_name, upload_id)
    except S3Error as exc:
        client.abort_multipart_upload(bucket_name, object_name, upload_id)
        raise exc

接口服务搭建

第3步：创建 FastAPI 上传路由

定义 POST 类型的 /upload/ 端点，接收前端传递的文件流。解析完成后调用封装好的分块上传方法，并返回标准化的 JSON 响应。建议在生产中替换硬编码的凭证与 Bucket 名称。

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    try:
        client = minio_client()
        # 使用 file.filename 作为目标对象名，避免底层文件句柄路径差异导致报错
        upload_file_with_resume(client, "my-bucketname", file.filename, file.file.name)
        return JSONResponse(status_code=200, content={"message": "File uploaded successfully"})
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})

前端调用示例

第4步：Axios 请求封装

浏览器端通过 FormData 组装文件数据，利用 Axios 发起多部分表单请求。监听输入框变化触发上传，并根据返回状态码或拦截器捕获进行结果反馈。

async function uploadFileToMinIO(file) {
  const formData = new FormData();
  formData.append('file', file);

  try {
    const response = await axios.post('http://localhost:8000/upload/', formData, {
      headers: {
        'Content-Type': 'multipart/form-data'
      }
    });
    console.log(response.data);
  } catch (error) {
    console.error('Error uploading file:', error);
  }
}

// 绑定 DOM 事件触发上传
const fileInput = document.getElementById('fileInput');
fileInput.addEventListener('change', async (event) => {
  const file = event.target.files[0];
  await uploadFileToMinIO(file);
});

关键注意事项

• 安全性：生产环境务必启用 HTTPS 传输，并将访问密钥集中管理或使用环境变量注入，严禁硬编码至源码仓库。
• 错误处理：建议将 S3Error 中的具体 HTTP 状态码映射到业务级错误提示，便于定位是分块失败还是权限不足。
• 性能优化：part_size 默认设为 10MB，可根据服务器带宽与存储节点负载调整该参数以平衡内存占用与并发效率。
• 断点续传完善：当前代码实现了标准的多分片上传架构。若要真正支持网络中断后“从中断点恢复”，需在循环中加入 client.list_parts() 检查已有分块 ID，跳过已上传部分并计算读取指针偏移量。

总结

本文提供了基于 Python 与 FastAPI 的 MinIO 分块上传完整链路方案。通过解耦客户端请求与底层对象存储交互，能够稳定承载大体积文件的上传任务。在实际落地时，结合偏移量校验即可补全完整的断点续传能力。以上就是基于Python和FastAPI实现MinIO分块上传与断点续传的详细内容，更多关于Python MinIO接口开发与前端集成的资料请关注本站其它相关文章！