如何在Flask微服务中实现简单的幂等POST /upload端点？

核心思路 #

基于PostgreSQL的ON CONFLICT语法结合RETURNING子句，配合数据库唯一约束，无需额外依赖就能满足需求：

• 先给photos表添加UNIQUE(user_id, filename)约束，从数据库层面阻断重复数据插入
• 修改插入语句，用ON CONFLICT DO NOTHING避免报错，同时通过RETURNING *获取插入成功的行记录
• 根据执行结果判断是首次请求还是重复请求，返回对应的HTTP状态码

完整代码实现 #

@app.post("/upload")
def upload():
    payload = request.get_json()
    # 执行带冲突处理和返回值的插入语句
    result = db.execute(
        """
        INSERT INTO photos (user_id, filename, uploaded_at)
        VALUES (%s, %s, NOW())
        ON CONFLICT (user_id, filename) DO NOTHING
        RETURNING *
        """,
        (payload["user_id"], payload["filename"]),
    )
    # 获取插入成功的行（首次请求有数据，重复请求为空）
    inserted_row = result.fetchone()
    
    if inserted_row:
        # 首次插入，返回201 Created
        return jsonify({"status": "ok", "message": "照片已成功上传"}), 201
    else:
        # 重复请求，返回200 OK
        return jsonify({"status": "ok", "message": "照片已存在"}), 200

关键细节说明 #

• 唯一约束前置：必须先在数据库执行ALTER TABLE photos ADD CONSTRAINT unique_user_filename UNIQUE(user_id, filename);，这是实现幂等性的基础
• ON CONFLICT DO NOTHING：遇到重复的user_id+filename组合时，不执行插入操作，也不会抛出SQL错误
• RETURNING *`：让数据库返回本次插入成功的行数据，重复请求时结果为空，以此区分请求类型
• 状态码规范：严格遵循HTTP语义，首次创建资源返回201，重复操作返回200，客户端可根据状态码做对应处理

方案优势 #

• 无额外依赖：仅使用Flask和PostgreSQL原生功能，无需引入第三方库或中间件
• 幂等性保障：重复提交不会产生重复数据，也不会触发错误
• 客户端反馈明确：不同场景返回不同状态码和提示信息，逻辑清晰易懂

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