Flask应用如何获取S3文件下载进度实现前端进度条展示

你当前的实现跑不通实时进度，核心是两个问题：

• Flask默认同步处理请求，要是把S3下载逻辑直接写在触发下载的POST接口里，整个接口会等文件全下完才返回响应，前端根本拿不到中间的进度数据
• OSBucket里的progressPercent、downloaded、_size都是实例绑定的变量，多用户同时下文件的时候会出现数据串扰，进度没法和具体用户的任务对应上

完整实现步骤如下：

第一步：改造后端逻辑 #

核心思路是给每个下载任务生成唯一ID，把下载任务扔到后台线程跑避免阻塞主请求，用线程安全的全局存储单独记录每个任务的进度、状态，再提供单独的进度查询接口给前端轮询。

1. 先导入依赖，初始化线程安全的任务存储

from flask import Flask, request, jsonify
import uuid
import threading
import os
import logging
from botocore.exceptions import ClientError
# 保留你原有S3客户端、OSBucket的初始化依赖

# 提前创建文件本地保存目录
os.makedirs('./downloads', exist_ok=True)
# 全局任务存储：key为任务ID，value存进度、状态、文件信息
task_store = {}
# 线程锁，避免多任务并发写存储时出现数据错乱
task_lock = threading.Lock()

2. 改造OSBucket类，让进度更新绑定到对应任务ID

class OSBucket():
    def __init__(self):
        # 保留你原有初始化逻辑，比如_s3客户端、BUCKET名配置
        self.downloaded = 0
        self._size = 0
        self.progressPercent = 0

    def download_callback(self, bytes_transferred, task_id):
        if self._size != 0:         
            self.downloaded += bytes_transferred
            self.progressPercent = int(self.downloaded / self._size * 100)
            # 实时更新对应任务的进度
            with task_lock:
                if task_id in task_store:
                    task_store[task_id]['progress'] = self.progressPercent
            print(f"任务{task_id}下载进度：{self.progressPercent}%")        
         
    def download_my_file(self, fileName, filepath, task_id):
        self._size = self._s3.head_object(Bucket=self.BUCKET, Key=fileName)['ContentLength']
        # 初始化任务状态
        with task_lock:
            task_store[task_id]['total_size'] = self._size
            task_store[task_id]['status'] = 'downloading'
        try:
            self._s3.download_file(
                Bucket=self.BUCKET, 
                Key=fileName, 
                Filename=filepath, 
                Callback=lambda bytes_transferred: self.download_callback(bytes_transferred, task_id)
            )
            # 下载完成更新状态
            with task_lock:
                task_store[task_id]['status'] = 'success'
                task_store[task_id]['filepath'] = filepath
        except ClientError as e:
            logging.error(e)
            with task_lock:
                task_store[task_id]['status'] = 'failed'
                task_store[task_id]['error'] = str(e)
            return False
        return True

3. 编写两个Flask接口，分别负责启动下载、查询进度

app = Flask(__name__)

# 触发下载接口
@app.route('/api/start-download', methods=['POST'])
def start_download():
    req_data = request.get_json()
    file_name = req_data.get('fileName')
    if not file_name:
        return jsonify({'code': 400, 'msg': '文件名不能为空'}), 400
    # 生成唯一任务ID
    task_id = str(uuid.uuid4())
    # 初始化任务信息
    with task_lock:
        task_store[task_id] = {
            'progress': 0,
            'status': 'pending',
            'fileName': file_name
        }
    # 下载逻辑放到后台线程跑，不阻塞当前请求
    def run_download():
        save_path = f"./downloads/{file_name}"
        # 每个任务单独初始化OSBucket实例，避免实例变量串扰
        bucket = OSBucket()
        bucket.download_my_file(file_name, save_path, task_id)
    threading.Thread(target=run_download, daemon=True).start()
    # 立刻返回任务ID给前端
    return jsonify({'code': 200, 'data': {'taskId': task_id}})

# 进度查询接口
@app.route('/api/download-progress/<task_id>', methods=['GET'])
def get_progress(task_id):
    with task_lock:
        task_info = task_store.get(task_id)
    if not task_info:
        return jsonify({'code': 404, 'msg': '任务不存在'}), 404
    return jsonify({'code': 200, 'data': task_info})

if __name__ == '__main__':
    # 开多线程模式，避免轮询请求被阻塞
    app.run(debug=True, threaded=True)

第二步：前端页面实现 #

逻辑很简单：点击下载按钮先调用启动接口拿到任务ID，之后开定时器每隔几百毫秒查询一次进度，更新进度条，等任务状态变为成功/失败时停掉定时器即可。

<div style="width: 500px; margin: 50px auto;">
    <input type="text" id="fileNameInput" placeholder="输入要下载的文件名" value="test.pdf">
    <button id="downloadBtn">开始下载</button>
    <div style="margin-top: 20px; border: 1px solid #ccc; height: 30px;">
        <div id="progressBar" style="height: 100%; width: 0%; background: #409eff; transition: width 0.2s;"></div>
    </div>
    <p id="progressText">等待开始下载</p>
</div>

<script>
const downloadBtn = document.getElementById('downloadBtn')
const progressBar = document.getElementById('progressBar')
const progressText = document.getElementById('progressText')
const fileNameInput = document.getElementById('fileNameInput')
let progressTimer = null

downloadBtn.addEventListener('click', async () => {
    const fileName = fileNameInput.value.trim()
    if(!fileName) {
        alert('请输入文件名')
        return
    }
    // 重置状态
    downloadBtn.disabled = true
    progressBar.style.width = '0%'
    progressText.innerText = '正在启动下载任务...'

    // 调用启动接口拿任务ID
    const startRes = await fetch('/api/start-download', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({fileName})
    }).then(res => res.json())
    if(startRes.code !== 200) {
        progressText.innerText = `启动失败：${startRes.msg}`
        downloadBtn.disabled = false
        return
    }
    const taskId = startRes.data.taskId

    // 轮询查询进度
    progressTimer = setInterval(async () => {
        const progressRes = await fetch(`/api/download-progress/${taskId}`).then(res => res.json())
        if(progressRes.code !== 200) {
            clearInterval(progressTimer)
            progressText.innerText = '查询进度失败'
            downloadBtn.disabled = false
            return
        }
        const {progress, status, error} = progressRes.data
        // 更新进度条
        progressBar.style.width = `${progress}%`
        progressText.innerText = `下载中：${progress}%`

        // 任务结束停止轮询
        if(status === 'success') {
            clearInterval(progressTimer)
            progressText.innerText = '下载完成！'
            downloadBtn.disabled = false
            // 后续可加文件保存到用户本地的逻辑，比如跳转文件地址、触发流下载
        } else if(status === 'failed') {
            clearInterval(progressTimer)
            progressText.innerText = `下载失败：${error}`
            downloadBtn.disabled = false
        }
    }, 300) // 300毫秒查询一次，平衡实时性和服务器压力
})
</script>

可选优化点 #

• 要是追求更高实时性、减少无效请求，可以把轮询换成SSE（服务器推送事件），后端进度更新后直接推给前端，不用反复发HTTP请求，更适合大文件下载场景
• 生产环境不要用内存字典存任务信息，服务重启会丢数据，可以换成Redis存储任务状态，同时支持多实例部署
• 加定时清理逻辑，自动删除下载完成超过1小时的旧任务数据，避免内存被占满
• 如果不需要把文件存在Flask服务本地，可以直接生成S3预签名URL返回给前端，让前端直连S3下载，省掉Flask服务的带宽开销

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