Python 大文件分片上传的处理方法-Python教程-PHP中文网

Python 大文件分片上传的处理方法

舞姬之光

发布： 2025-11-23 23:31:02

原创

838人浏览过

大文件分片上传通过切块避免内存溢出，客户端逐片上传并携带元数据，服务端按序合并；核心是生成器读取、分片传输与完整校验，支持断点续传和进度跟踪，确保GB级文件稳定上传。

python 大文件分片上传的处理方法

大文件上传在Web开发中很常见，直接上传可能因内存占用高或网络中断导致失败。Python中实现大文件分片上传，核心思路是将文件切块、逐个上传、服务端合并。以下是具体处理方法。

1. 文件分片读取

避免一次性加载整个文件到内存，使用生成器按固定大小读取文件块。

设定分片大小（如5MB）
通过循环读取文件流，每次返回一部分数据
适合任意大小的文件，内存占用稳定

示例代码：

def read_file_chunks(file_path, chunk_size=5 * 1024 * 1024):
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

登录后复制

2. 客户端上传分片

每上传一个分片，携带必要信息：文件名、分片序号、总分片数等，便于服务端识别和重组。

立即学习“Python免费学习笔记（深入）”；

使用requests库发送POST请求
每个请求附带当前分片数据及元信息（可放在form-data中）
支持断点续传时，可先查询已上传的分片

上传示例：

Metronic Bootstrap后台模板

Metronic是一套精美的响应式后台管理模板，基于强大的Twitter Bootstrap框架实现。Metronic拥有简洁优雅的Metro UI风格界面，自适应屏幕分辨率大小，兼容PC端和手机移动端。全套模板，包含仪表盘、侧边栏菜单、布局宣传片、电子邮件模板、UI特性、按钮、标签、表格布局、表单组件、多文件上传、悬浮窗文件上传、时间表、博客、新闻、关于我们、联系我们、日历、用户配置文件、锁屏、

275

查看详情

import requests
<p>for index, chunk in enumerate(read_file_chunks('large<em>file.zip')):
files = {'file': ('chunk</em>%d' % index, chunk)}
data = {
'filename': 'large_file.zip',
'chunk_index': index,
'total_chunks': 10  # 可预先计算
}
response = requests.post('<a href="https://www.php.cn/link/c0dba5809c620f70942856ad09b144d0">https://www.php.cn/link/c0dba5809c620f70942856ad09b144d0</a>', data=data, files=files)
if response.status_code != 200:
print(f"上传失败: 分片 {index}")
break

登录后复制

3. 服务端接收与合并

服务端需保存每个分片，待全部接收后按顺序合并。

根据文件名创建临时目录存放分片
收到分片后以序号命名存储（如 part_0, part_1）
检查是否所有分片已上传，自动触发合并

Flask 示例处理逻辑：

from flask import Flask, request
import os
<p>app = Flask(<strong>name</strong>)
UPLOAD_DIR = '/tmp/uploads'
CHUNKS_DIR = '/tmp/chunks'</p><p>@app.route('/upload', methods=['POST'])
def handle_upload():
file = request.files['file']
filename = request.form['filename']
chunk_index = int(request.form['chunk_index'])
total_chunks = int(request.form['total_chunks'])</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;">chunk_dir = os.path.join(CHUNKS_DIR, filename + "_parts")
os.makedirs(chunk_dir, exist_ok=True)

chunk_path = os.path.join(chunk_dir, f"part_{chunk_index}")
file.save(chunk_path)

# 检查是否全部上传完成
if all(os.path.exists(os.path.join(chunk_dir, f"part_{i}")) for i in range(total_chunks)):
    merge_files(chunk_dir, os.path.join(UPLOAD_DIR, filename))
    cleanup(chunk_dir)  # 删除分片

return "OK", 200

登录后复制

def merge_files(chunk_dir, target_path): with open(target_path, 'wb') as f: for i in sorted(os.listdir(chunk_dir)): part_path = os.path.join(chunk_dir, i) with open(part_path, 'rb') as part: f.write(part.read())