如何基于Python Flask+Resumable.js实现文件上传信息与状态的数据库记录

我来给你梳理一套实战可行的实现方案，从数据库设计、后端核心逻辑到状态跟踪全覆盖，都是实际项目中验证过的思路。

一、数据库表设计（以SQLAlchemy为例） #

首先得建一张表来存储文件的元数据和上传状态，字段要覆盖你需要的所有信息：

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import uuid

db = SQLAlchemy()

class UploadedFile(db.Model):
    id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    filename = db.Column(db.String(255), nullable=False)  # 原始文件名
    file_size = db.Column(db.BigInteger, nullable=False)  # 文件总大小（字节）
    last_modified = db.Column(db.DateTime, nullable=False)  # 文件修改时间
    status = db.Column(db.String(20), nullable=False, default='uploading')  # 状态：uploading/completed/canceled/paused
    total_chunks = db.Column(db.Integer, nullable=False)  # 总分片数
    uploaded_chunks = db.Column(db.Integer, default=0)  # 已上传分片数
    storage_path = db.Column(db.String(500))  # 最终文件存储路径
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

字段说明：

• id用UUID作为唯一标识，避免文件名冲突，也方便前后端关联
• status字段是核心，用来跟踪上传状态：uploading（上传中）、completed（完成）、canceled（已取消）、paused（已暂停）
• last_modified可以直接存前端传过来的时间戳转成的DateTime对象

二、Flask后端核心接口实现 #

Resumable.js的工作流程是分片上传，前端会先检查分片是否已上传，再上传未完成的分片，最后合并。我们需要对应实现几个关键接口：

1. 初始化文件记录接口（前端上传前调用） #

前端在开始上传前，先把文件的元数据传给后端，创建一条上传中的记录：

from flask import Flask, request, jsonify
import os

app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = '/path/to/your/upload/directory'
app.config['TEMP_FOLDER'] = '/path/to/your/temp/directory'
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://user:password@localhost/db_name'

# 确保目录存在
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
os.makedirs(app.config['TEMP_FOLDER'], exist_ok=True)

@app.route('/api/upload/init', methods=['POST'])
def init_upload():
    data = request.get_json()
    filename = data['filename']
    file_size = data['file_size']
    last_modified = datetime.fromtimestamp(data['last_modified'] / 1000)  # 前端传的是毫秒时间戳
    total_chunks = data['total_chunks']

    # 创建文件记录
    upload_file = UploadedFile(
        filename=filename,
        file_size=file_size,
        last_modified=last_modified,
        total_chunks=total_chunks
    )
    db.session.add(upload_file)
    db.session.commit()

    return jsonify({'file_id': upload_file.id}), 201

2. 分片存在性检查接口 #

Resumable.js会先请求这个接口，确认哪些分片已经上传过，避免重复上传：

@app.route('/api/upload/check', methods=['GET'])
def check_chunk():
    file_id = request.args.get('file_id')
    chunk_index = request.args.get('chunk_index', type=int)

    # 检查数据库中该文件的已上传分片是否包含当前分片
    upload_file = UploadedFile.query.get(file_id)
    if not upload_file:
        return jsonify({'exists': False})
    
    # 也可以检查临时目录中是否存在该分片文件（双重验证更安全）
    chunk_path = os.path.join(app.config['TEMP_FOLDER'], f"{file_id}_chunk_{chunk_index}")
    exists = os.path.exists(chunk_path) and (upload_file.uploaded_chunks >= chunk_index + 1)

    return jsonify({'exists': exists})

3. 分片上传接口 #

接收前端传来的分片，保存到临时目录，并更新数据库的已上传分片数：

@app.route('/api/upload/chunk', methods=['POST'])
def upload_chunk():
    file_id = request.form.get('file_id')
    chunk_index = request.form.get('chunk_index', type=int)
    chunk_file = request.files.get('file')

    upload_file = UploadedFile.query.get(file_id)
    if not upload_file:
        return jsonify({'error': 'File record not found'}), 404
    
    # 如果文件状态是已取消或已完成，拒绝上传
    if upload_file.status in ['canceled', 'completed']:
        return jsonify({'error': 'Upload is canceled or completed'}), 403

    # 保存分片到临时目录
    chunk_path = os.path.join(app.config['TEMP_FOLDER'], f"{file_id}_chunk_{chunk_index}")
    chunk_file.save(chunk_path)

    # 更新已上传分片数（这里可以加锁，避免并发上传导致计数错误）
    upload_file.uploaded_chunks = db.func.coalesce(upload_file.uploaded_chunks, 0) + 1
    db.session.commit()

    # 检查是否所有分片都已上传
    if upload_file.uploaded_chunks == upload_file.total_chunks:
        return jsonify({'status': 'all_chunks_uploaded', 'file_id': file_id})
    else:
        return jsonify({'status': 'chunk_uploaded', 'uploaded_chunks': upload_file.uploaded_chunks})

4. 合并分片接口 #

当所有分片上传完成后，前端调用此接口合并成完整文件：

def merge_chunks(file_id, filename):
    upload_file = UploadedFile.query.get(file_id)
    # 生成唯一的存储文件名，避免重名
    storage_filename = f"{file_id}_{filename}"
    storage_path = os.path.join(app.config['UPLOAD_FOLDER'], storage_filename)

    # 合并所有分片
    with open(storage_path, 'wb') as dest_file:
        for i in range(upload_file.total_chunks):
            chunk_path = os.path.join(app.config['TEMP_FOLDER'], f"{file_id}_chunk_{i}")
            with open(chunk_path, 'rb') as chunk_file:
                dest_file.write(chunk_file.read())
            # 删除临时分片文件
            os.remove(chunk_path)
    
    return storage_path

@app.route('/api/upload/merge', methods=['POST'])
def merge_file():
    data = request.get_json()
    file_id = data['file_id']

    upload_file = UploadedFile.query.get(file_id)
    if not upload_file:
        return jsonify({'error': 'File record not found'}), 404
    
    if upload_file.status != 'uploading':
        return jsonify({'error': 'Invalid upload status'}), 403
    
    try:
        storage_path = merge_chunks(file_id, upload_file.filename)
        # 更新数据库记录
        upload_file.status = 'completed'
        upload_file.storage_path = storage_path
        db.session.commit()
        return jsonify({'status': 'completed', 'storage_path': storage_path})
    except Exception as e:
        db.session.rollback()
        return jsonify({'error': str(e)}), 500

5. 取消上传接口 #

前端点击取消按钮时调用，更新状态并清理临时文件：

@app.route('/api/upload/cancel', methods=['POST'])
def cancel_upload():
    data = request.get_json()
    file_id = data['file_id']

    upload_file = UploadedFile.query.get(file_id)
    if not upload_file:
        return jsonify({'error': 'File record not found'}), 404
    
    upload_file.status = 'canceled'
    db.session.commit()

    # 清理临时分片文件
    for i in range(upload_file.total_chunks):
        chunk_path = os.path.join(app.config['TEMP_FOLDER'], f"{file_id}_chunk_{i}")
        if os.path.exists(chunk_path):
            os.remove(chunk_path)
    
    return jsonify({'status': 'canceled'})

6. 暂停上传接口 #

前端点击暂停时调用，仅更新状态（不需要清理文件，方便后续恢复）：

@app.route('/api/upload/pause', methods=['POST'])
def pause_upload():
    data = request.get_json()
    file_id = data['file_id']

    upload_file = UploadedFile.query.get(file_id)
    if not upload_file:
        return jsonify({'error': 'File record not found'}), 404
    
    if upload_file.status == 'uploading':
        upload_file.status = 'paused'
        db.session.commit()
        return jsonify({'status': 'paused'})
    else:
        return jsonify({'error': 'Invalid upload status'}), 403

三、状态跟踪的补充说明 #

1. 恢复上传：当用户暂停后再次上传，前端可以通过file_id查询数据库，获取已上传分片数，然后从下一个分片开始上传，后端的分片检查接口会自动跳过已上传的分片。
2. 定时清理：可以写一个定时任务（比如用APScheduler），定期清理状态为canceled且超过24小时的记录，以及对应的临时文件；对于paused状态的记录，可以保留7天再清理。
3. 前端配合：前端需要在Resumable.js的事件回调中调用对应的接口，比如：
   • 开始上传前调用/api/upload/init获取file_id
   • 暂停时调用/api/upload/pause
   • 取消时调用/api/upload/cancel
   • 所有分片上传完成后调用/api/upload/merge

四、注意事项 #

• 并发处理：分片上传可能存在并发请求，更新uploaded_chunks时建议用数据库的原子操作（比如上面的db.func.coalesce），避免计数错误。
• 文件安全：存储文件名用UUID+原始文件名的组合，避免恶意文件名导致的路径遍历问题；同时限制上传文件的类型和大小。
• 事务处理：在合并分片等关键操作中，要使用数据库事务，避免出现数据不一致的情况。

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