如何在Python的RESTful API中中断shutil.copyfile的大文件复制进程？

嘿，我来帮你搞定这个问题～你现在用的shutil.copyfile有个小局限：它是一次性把整个文件读入内存再写入，全程阻塞，一旦开始复制大文件，就没法中途停下——这也是为啥你没法中断特定API调用的原因。要解决这个，我们得换个思路：分块复制，每复制一小段就检查一次是否需要中断，这样既能及时响应取消请求，又不会影响服务本身。

第一步：写一个支持中断的分块复制函数 #

我们自己实现分块复制逻辑，代替shutil.copyfile，核心是在每块复制的间隙加入中断检查。代码如下：

import os

def copy_file_interruptible(source, destination, check_interrupt):
    """支持中断的分块复制函数
    check_interrupt: 无参回调函数，返回True时表示要中断复制
    """
    chunk_size = 1024 * 1024  # 用1MB的块来复制，你可以根据服务器性能调整大小
    # 先确保目标目录存在
    dest_dir = os.path.dirname(destination)
    if not os.path.exists(dest_dir):
        os.makedirs(dest_dir)
    
    try:
        with open(source, 'rb') as src_file, open(destination, 'wb') as dest_file:
            while True:
                # 每次复制前先检查是否需要中断
                if check_interrupt():
                    # 中断时记得删掉未完成的目标文件，避免留垃圾
                    os.remove(destination)
                    return False
                # 读取一块内容
                chunk = src_file.read(chunk_size)
                if not chunk:
                    break  # 读完了，复制完成
                # 写入一块内容
                dest_file.write(chunk)
        return True
    except Exception as e:
        # 出错时也清理未完成的文件
        if os.path.exists(destination):
            os.remove(destination)
        raise e

这个函数的关键是check_interrupt回调——我们可以在里面判断当前的API请求是不是被用户中断了。

第二步：在RESTful API里集成中断检测 #

假设你用的是Flask（其他框架比如Django逻辑类似，核心都是检测请求是否断开），Flask可以通过request.environ里的werkzeug.request_disconnected来判断客户端是不是断开了连接。我们把这个逻辑封装成检查函数传给上面的复制函数：

from flask import request, Flask, jsonify

app = Flask(__name__)

def check_if_request_disconnected():
    """检查当前请求是否被客户端取消/断开"""
    return request.environ.get('werkzeug.request_disconnected', False)

@app.route('/copy', methods=['POST'])
def copy_file_api():
    # 从请求里拿源文件和目标路径（这里假设用JSON传参）
    source = request.json.get('source')
    destination = request.json.get('destination')
    
    if not source or not destination:
        return jsonify({'status': 'error', 'message': '缺少source或destination参数'}), 400
    
    try:
        # 调用可中断的复制函数
        copy_success = copy_file_interruptible(source, destination, check_if_request_disconnected)
        if copy_success:
            return jsonify({'status': 'success', 'message': '文件复制完成'})
        else:
            return jsonify({'status': 'cancelled', 'message': '复制操作已被取消'}), 409
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)}), 500

如果是Django的话，你可以通过request对象的连接状态来判断，或者结合前端的取消信号（比如AJAX的abort事件），核心都是在分块复制的间隙判断是否要终止操作。

额外优化点 #

• 块大小调整：chunk_size可以根据你的服务器性能和文件大小调，太大的话中断响应会变慢，太小会增加IO次数，1MB是个比较均衡的选择。
• 后台线程处理：如果你的API是同步阻塞的，大文件复制会占用请求线程，导致其他请求排队。这时候可以把复制操作放到后台线程里，同时用一个线程安全的标志位来控制中断，比如：

import threading
from flask import request, Flask, jsonify

app = Flask(__name__)
# 用字典存每个线程的中断标志，key用线程ID
interrupt_flags = {}

def copy_in_background(source, destination, thread_id):
    chunk_size = 1024 * 1024
    dest_dir = os.path.dirname(destination)
    if not os.path.exists(dest_dir):
        os.makedirs(dest_dir)
    
    try:
        with open(source, 'rb') as src_file, open(destination, 'wb') as dest_file:
            while True:
                # 检查当前线程的中断标志
                if interrupt_flags.get(thread_id, False):
                    os.remove(destination)
                    break
                chunk = src_file.read(chunk_size)
                if not chunk:
                    break
                dest_file.write(chunk)
    except Exception as e:
        if os.path.exists(destination):
            os.remove(destination)
        print(f"复制失败: {e}")
    finally:
        # 清理标志位
        interrupt_flags.pop(thread_id, None)

@app.route('/copy', methods=['POST'])
def start_copy():
    source = request.json.get('source')
    destination = request.json.get('destination')
    if not source or not destination:
        return jsonify({'status': 'error', 'message': '缺少必要参数'}), 400
    
    # 启动后台线程执行复制
    current_thread_id = threading.get_ident()
    interrupt_flags[current_thread_id] = False
    thread = threading.Thread(target=copy_in_background, args=(source, destination, current_thread_id))
    thread.start()
    
    return jsonify({'status': 'started', 'thread_id': current_thread_id})

@app.route('/copy/cancel/<int:thread_id>', methods=['POST'])
def cancel_copy(thread_id):
    # 设置对应线程的中断标志
    interrupt_flags[thread_id] = True
    return jsonify({'status': 'success', 'message': '已请求取消复制'})

这个版本允许用户通过调用/copy/cancel/<thread_id>来主动中断复制，适合需要用户手动取消的场景。

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