Python防重提交案例如何防止重复提交

本文目录导读：

1. 使用Token机制（最常用）
2. Redis分布式锁方案（适合分布式系统）
3. 使用数据库唯一约束
4. 前端+后端综合方案（最推荐）
5. 最佳实践建议

防止重复提交是Web开发中非常常见的需求,我来分享几种实用的Python防重提交方案。

使用Token机制（最常用）

后端实现（Flask/Django）

# forms.py
import uuid
from flask import session, jsonify, request
from functools import wraps
# 生成Token
def generate_form_token():
    """生成防重提交Token"""
    token = uuid.uuid4().hex
    if 'form_tokens' not in session:
        session['form_tokens'] = []
    session['form_tokens'].append(token)
    # 限制Token数量，防止session过大
    if len(session['form_tokens']) > 10:
        session['form_tokens'] = session['form_tokens'][-10:]
    return token
# 验证Token装饰器
def validate_token(view_func):
    @wraps(view_func)
    def wrapper(*args, **kwargs):
        token = request.form.get('_token') or request.json.get('_token')
        if not token:
            return jsonify({'error': '缺少安全Token'}), 400
        if token not in session.get('form_tokens', []):
            return jsonify({'error': '表单已提交或Token已过期'}), 400
        # 移除已使用的Token
        session['form_tokens'].remove(token)
        return view_func(*args, **kwargs)
    return wrapper
# 使用示例
@app.route('/form_page')
def form_page():
    token = generate_form_token()
    return render_template('form.html', token=token)
@app.route('/submit', methods=['POST'])
@validate_token
def submit_form():
    # 处理表单逻辑
    return jsonify({'message': '提交成功'})

前端实现

<!-- form.html -->
<form id="myForm" action="/submit" method="POST">
    <input type="hidden" name="_token" value="{{ token }}">
    <input type="text" name="username" required>
    <button type="submit" id="submitBtn">提交</button>
</form>
<script>
document.getElementById('myForm').addEventListener('submit', function(e) {
    const submitBtn = document.getElementById('submitBtn');
    submitBtn.disabled = true;  // 按钮置灰
    submitBtn.textContent = '提交中...';
    // 还可以用一个标志位防止连续点击
});
</script>

Redis分布式锁方案（适合分布式系统）

# redis_lock.py
import redis
import time
import uuid
from functools import wraps
class RedisLock:
    """基于Redis的分布式锁"""
    def __init__(self, redis_client, lock_key, expire=5):
        self.redis = redis_client
        self.lock_key = f"lock:{lock_key}"
        self.expire = expire
        self.owner = str(uuid.uuid4())
    def acquire(self, timeout=0):
        """获取锁"""
        end = time.time() + timeout
        while timeout == 0 or time.time() < end:
            # SETNX + EXPIRE 原子操作
            acquired = self.redis.setnx(self.lock_key, self.owner)
            if acquired:
                self.redis.expire(self.lock_key, self.expire)
                return True
            if timeout == 0:
                return False
            time.sleep(0.1)
        return False
    def release(self):
        """释放锁 - 只释放自己持有的锁"""
        # Lua脚本保证原子性
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        self.redis.eval(script, 1, self.lock_key, self.owner)
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()
# 使用Redis锁防重提交
class PreventDuplicateSubmit:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
    def prevent_duplicate(self, key_func=None, expire=5):
        """装饰器，防止重复提交"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 生成唯一锁键
                if key_func:
                    lock_key = key_func(*args, **kwargs)
                else:
                    # 默认使用请求参数生成键
                    request = kwargs.get('request') or args[0]
                    lock_key = f"{func.__name__}:{request.remote_addr}:{request.form.get('id', '')}"
                lock = RedisLock(self.redis, lock_key, expire=expire)
                if lock.acquire(timeout=0):
                    try:
                        return func(*args, **kwargs)
                    finally:
                        lock.release()
                else:
                    return {"error": "操作正在进行中，请勿重复提交"}, 429
            return wrapper
        return decorator
# 使用示例
prevent = PreventDuplicateSubmit()
@app.route('/order/create', methods=['POST'])
@prevent.prevent_duplicate(key_func=lambda req: f"order:{req.form.get('product_id')}:{req.remote_addr}")
def create_order():
    # 创建订单逻辑
    pass

使用数据库唯一约束

# model_unique.py
from datetime import datetime
from sqlalchemy import Column, String, Integer, DateTime, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.exc import IntegrityError
Base = declarative_base()
class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    order_no = Column(String(64), unique=True)  # 唯一约束
    user_id = Column(Integer)
    product_id = Column(Integer)
    amount = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)
    __table_args__ = (
        UniqueConstraint('user_id', 'product_id', name='user_product_uk'),
    )
# 防重提交服务
class PreventDuplicateService:
    def __init__(self, session_factory):
        self.Session = session_factory
    @staticmethod
    def generate_order_no(user_id, product_id):
        """生成业务唯一标识"""
        timestamp = int(time.time() * 1000)
        return f"ORD{user_id}{product_id}{timestamp}"
    def create_order(self, user_id, product_id, amount):
        """幂等创建订单"""
        session = self.Session()
        try:
            # 先检查是否存在
            existing = session.query(Order).filter_by(
                user_id=user_id, 
                product_id=product_id
            ).first()
            if existing:
                return {
                    'success': False,
                    'message': '该商品已下单',
                    'data': existing
                }
            # 创建订单
            order = Order(
                order_no=self.generate_order_no(user_id, product_id),
                user_id=user_id,
                product_id=product_id,
                amount=amount
            )
            session.add(order)
            session.commit()
            return {
                'success': True,
                'message': '订单创建成功',
                'data': order
            }
        except IntegrityError:
            session.rollback()
            return {
                'success': False,
                'message': '订单已存在，请勿重复提交'
            }
        finally:
            session.close()

前端+后端综合方案（最推荐）

# comprehensive_prevent.py
import hashlib
import json
import time
from functools import wraps
class ComprehensiveAntiDuplicate:
    """综合防重提交方案"""
    def __init__(self, cache_client):
        self.cache = cache_client  # 可以是Redis, Memcached等
    def _generate_request_fingerprint(self, request):
        """生成请求指纹"""
        data = {
            'url': request.path,
            'method': request.method,
            'body': request.get_data(as_text=True),
            'timestamp': int(time.time() // 2),  # 2秒内的请求视为重复
            'user': request.remote_addr
        }
        return hashlib.md5(json.dumps(data, sort_keys=True).encode()).hexdigest()
    def anti_duplicate(self, expire=3):
        """防重提交装饰器"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                request = kwargs.get('request') or args[0]
                # 1. 前端防重（检查Token）
                client_token = request.headers.get('X-Request-Id') or request.form.get('_request_id')
                if not client_token:
                    return {'error': '缺少请求标识'}, 400
                # 2. 后端防重（检查请求指纹）
                fingerprint = self._generate_request_fingerprint(request)
                cache_key = f"duplicate:{fingerprint}"
                if self.cache.get(cache_key):
                    # 检查是否是幂等请求（允许重试）
                    if request.method in ['GET', 'HEAD', 'OPTIONS']:
                        pass  # 幂等方法允许重复
                    else:
                        return {'error': '请求正在处理或已处理，请勿重复提交'}, 429
                # 3. 设置缓存标记
                self.cache.setex(cache_key, expire, client_token)
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    # 请求失败时清理缓存
                    self.cache.delete(cache_key)
                    raise e
            return wrapper
        return decorator
# 5. 前端完整防重提交JavaScript

// anti_duplicate.js
class FormAntiDuplicate {
    constructor() {
        this.submitted = new Set();
        this.submitLock = false;
    }
    // 生成唯一请求ID
    generateRequestId() {
        return `req_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
    // 添加请求到已提交列表
    addSubmitted(formId) {
        this.submitted.add(formId);
    }
    // 检查是否已提交
    isSubmitted(formId) {
        return this.submitted.has(formId);
    }
    // 移除请求
    removeSubmitted(formId) {
        this.submitted.delete(formId);
    }
    // 防重提交的表单提交方法
    async submitForm(formElement, submitUrl) {
        const formId = formElement.id || formElement.name;
        // 1. 检查是否已提交
        if (this.isSubmitted(formId)) {
            console.error('表单已提交，请勿重复操作');
            return false;
        }
        // 2. 检查是否正在提交
        if (this.submitLock) {
            console.error('正在提交中，请稍候');
            return false;
        }
        // 3. 设置锁定状态
        this.submitLock = true;
        this.addSubmitted(formId);
        try {
            // 4. 添加请求标识
            const requestId = this.generateRequestId();
            const formData = new FormData(formElement);
            formData.append('_request_id', requestId);
            // 5. 发送请求
            const response = await fetch(submitUrl, {
                method: 'POST',
                body: formData,
                headers: {
                    'X-Request-Id': requestId
                }
            });
            const result = await response.json();
            if (!response.ok) {
                throw new Error(result.error || '提交失败');
            }
            // 6. 提交成功
            this.removeSubmitted(formId);
            return result;
        } catch (error) {
            // 7. 提交失败
            this.removeSubmitted(formId);
            throw error;
        } finally {
            // 8. 释放锁定
            this.submitLock = false;
        }
    }
}
// 使用示例
const antiDuplicate = new FormAntiDuplicate();
document.getElementById('submitBtn').addEventListener('click', async function(e) {
    e.preventDefault();
    const form = document.getElementById('myForm');
    const submitBtn = this;
    try {
        // 禁用按钮
        submitBtn.disabled = true;
        submitBtn.textContent = '提交中...';
        // 防重提交
        const result = await antiDuplicate.submitForm(form, '/api/submit');
        if (result) {
            alert('提交成功！');
        }
    } catch (error) {
        alert(error.message);
    } finally {
        // 恢复按钮
        submitBtn.disabled = false;
        submitBtn.textContent = '提交';
    }
});

最佳实践建议

1. 多层防御：前端+后端+数据库三层防重
2. 幂等设计：接口设计为幂等的，允许重复请求
3. 合理超时：设置合理的锁超时时间（通常3-5秒）
4. 异常处理：处理好网络超时、服务异常等情况
5. 日志记录：记录重复提交日志便于排查
6. 用户提示：友好的用户提示，不要直接显示技术错误

选择哪种方案取决于你的具体需求：

• 单机应用：Token机制就够了
• 分布式系统：Redis分布式锁
• 高并发场景：数据库唯一约束+幂等设计
• 最佳实践：综合使用多个方案