Flask-SQLAlchemy 数据重复插入问题及解决方案

聖光之護

发布时间：2025-11-20 14:12:06

325人浏览过

来源于php中文网

原创

Flask-SQLAlchemy 数据重复插入问题及解决方案

本文旨在探讨并解决在使用 flask 和 sqlalchemy 进行数据持久化时，由页面刷新或脚本重复执行导致的数据库数据重复插入问题。我们将深入分析两种核心策略：通过数据库层面的唯一性约束来阻止重复数据，以及利用 web 开发中的 post-redirect-get 模式来避免客户端意外的重复提交，同时也会解释 flask 应用上下文的正确使用。

引言：Flask-SQLAlchemy 数据重复插入的挑战

在使用 Flask 结合 SQLAlchemy 管理数据库时，开发者常会遇到一个常见但令人困扰的问题：数据在不经意间被重复插入数据库。这通常发生在以下场景：

页面刷新/回退操作： 用户在提交表单后刷新页面，或使用浏览器回退功能，可能导致 POST 请求被重新发送，从而触发数据再次插入。
脚本重复执行： 在开发或测试阶段，如果用于初始化或填充数据的脚本被多次运行，而没有适当的去重机制，也会导致数据重复。
缺乏业务逻辑校验： 在数据插入前，未对现有数据进行检查，导致即使数据已存在，仍会尝试插入新的重复记录。

原始代码示例展示了在 app.app_context() 中循环插入项目和经验数据，如果这段代码被多次执行，例如在每次应用启动时，或者在某个视图函数中未正确处理请求生命周期，就会导致数据库中出现大量重复记录。

# 原始数据插入逻辑示例
with app.app_context():
    for project_data_item in projectData:
        project_entry = Project(
            projectName=project_data_item["projectName"],
            projectDescription=project_data_item["projectDescription"],
            projectUrl=project_data_item["projectUrl"],
        )
        db.session.add(project_entry)

    for experience_data_item in experience_data:
        experience_entry = Experience(
            companyName=experience_data_item["companyName"],
            companyDescription=experience_data_item["companyDescription"],
            companyUrl=experience_data_item["companyUrl"],
            companyRole=experience_data_item["companyRole"],
            companyDuration=experience_data_item["companyDuration"],
            companyLocation=experience_data_item["companyLocation"],
            companyResponsibilities=experience_data_item["companyResponsibilities"],
            projects=experience_data_item["projects"]
        )
        db.session.add(experience_entry)

    db.session.commit()

接下来，我们将探讨两种主要的解决方案来有效防止此类重复插入。

解决方案一：通过数据库唯一性约束防止重复

最直接且可靠的方法是在数据库层面强制执行数据的唯一性。这意味着即使应用程序尝试插入重复数据，数据库也会拒绝该操作，并抛出错误。

1. 定义唯一性约束

在 SQLAlchemy 模型中，可以通过 unique=True 参数为单个字段设置唯一性，或者使用 UniqueConstraint 为多个字段组合设置唯一性。

示例：为 Project 模型设置 projectName 唯一

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import UniqueConstraint

db = SQLAlchemy()

class Project(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    projectName = db.Column(db.String(80), unique=True, nullable=False)
    projectDescription = db.Column(db.Text, nullable=True)
    projectUrl = db.Column(db.String(200), nullable=True)

    def __repr__(self):
        return f''

示例：为 Experience 模型设置多字段组合唯一

假设一个“经验”由公司名称、角色和持续时间共同确定其唯一性。

class Experience(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    companyName = db.Column(db.String(120), nullable=False)
    companyDescription = db.Column(db.Text, nullable=True)
    companyUrl = db.Column(db.String(200), nullable=True)
    companyRole = db.Column(db.String(120), nullable=False)
    companyDuration = db.Column(db.String(120), nullable=False)
    companyLocation = db.Column(db.String(120), nullable=True)
    companyResponsibilities = db.Column(db.Text, nullable=True)
    # projects 字段可能需要定义关系，此处省略

    __table_args__ = (
        UniqueConstraint('companyName', 'companyRole', 'companyDuration', name='_company_role_duration_uc'),
    )

    def __repr__(self):
        return f''

2. 处理唯一性约束冲突

当尝试插入违反唯一性约束的数据时，数据库会抛出 IntegrityError。在应用程序中，需要捕获并处理这个错误。

方法一：先查询后插入 (Select-Before-Insert)

在插入数据前，先查询数据库中是否已存在具有相同唯一标识的数据。

花生AI

B站推出的AI视频创作工具

下载

from sqlalchemy.exc import IntegrityError

def add_project_if_not_exists(project_data_item):
    with app.app_context():
        existing_project = Project.query.filter_by(projectName=project_data_item["projectName"]).first()
        if existing_project:
            print(f"项目 '{project_data_item['projectName']}' 已存在，跳过插入。")
            return existing_project
        else:
            project_entry = Project(
                projectName=project_data_item["projectName"],
                projectDescription=project_data_item["projectDescription"],
                projectUrl=project_data_item["projectUrl"],
            )
            db.session.add(project_entry)
            try:
                db.session.commit()
                print(f"项目 '{project_data_item['projectName']}' 插入成功。")
                return project_entry
            except IntegrityError:
                db.session.rollback() # 回滚事务
                print(f"项目 '{project_data_item['projectName']}' 插入失败，可能存在并发冲突。")
                # 重新尝试查询或处理
                return Project.query.filter_by(projectName=project_data_item["projectName"]).first()

# 使用示例
# for project_data_item in projectData:
#     add_project_if_not_exists(project_data_item)

方法二：尝试插入并捕获错误

直接尝试插入，如果发生 IntegrityError 则回滚事务。这种方法在并发量高时可能效率更高，因为它避免了额外的 SELECT 查询。

from sqlalchemy.exc import IntegrityError

def add_experience_robustly(experience_data_item):
    with app.app_context():
        experience_entry = Experience(
            companyName=experience_data_item["companyName"],
            companyDescription=experience_data_item["companyDescription"],
            companyUrl=experience_data_item["companyUrl"],
            companyRole=experience_data_item["companyRole"],
            companyDuration=experience_data_item["companyDuration"],
            companyLocation=experience_data_item["companyLocation"],
            companyResponsibilities=experience_data_item["companyResponsibilities"],
        )
        db.session.add(experience_entry)
        try:
            db.session.commit()
            print(f"经验 '{experience_data_item['companyName']} - {experience_data_item['companyRole']}' 插入成功。")
            return experience_entry
        except IntegrityError:
            db.session.rollback() # 回滚事务
            print(f"经验 '{experience_data_item['companyName']} - {experience_data_item['companyRole']}' 已存在或插入失败。")
            # 如果需要获取已存在的对象，可以在这里再次查询
            return Experience.query.filter_by(
                companyName=experience_data_item["companyName"],
                companyRole=experience_data_item["companyRole"],
                companyDuration=experience_data_item["companyDuration"]
            ).first()

# 使用示例
# for experience_data_item in experience_data:
#     add_experience_robustly(experience_data_item)

解决方案二：利用 POST-Redirect-GET 模式防止重复提交

在 Web 应用中，当用户提交表单（通常是 POST 请求）后，如果直接渲染页面或返回数据，用户刷新浏览器可能会导致浏览器重新发送 POST 请求，从而引发重复提交。POST-Redirect-GET (PRG) 模式是解决此问题的标准实践。

1. POST-Redirect-GET 模式原理

POST 请求： 用户通过表单提交数据到服务器（POST 请求）。
处理数据： 服务器接收并处理数据（例如，将数据存入数据库）。
重定向： 服务器不直接返回响应页面，而是发送一个重定向（HTTP 302 Found）响应到另一个 URL，通常是一个 GET 请求的页面。
GET 请求： 浏览器接收到重定向后，会发起一个新的 GET 请求到指定的 URL。
渲染页面： 服务器处理 GET 请求并返回页面。

通过这种方式，即使用户刷新浏览器，也只会刷新最后的 GET 请求页面，而不会重新发送 POST 请求。

2. Flask 中的实现

from flask import Flask, request, redirect, url_for, render_template_string
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# 假设 Project 模型已定义并包含 unique=True 的 projectName
class Project(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    projectName = db.Column(db.String(80), unique=True, nullable=False)
    projectDescription = db.Column(db.Text, nullable=True)
    projectUrl = db.Column(db.String(200), nullable=True)

    def __repr__(self):
        return f''

# 在应用上下文外创建表 (通常在独立脚本或第一次运行时)
with app.app_context():
    db.create_all()

@app.route('/', methods=['GET'])
def index():
    # 显示一个简单的表单
    return render_template_string("""
        添加新项目
        
            项目名称: 

            项目描述: 

            项目URL: 

            
        
        现有项目
        
        {% for project in projects %}
            {{ project.projectName }} - {{ project.projectDescription }}
        {% endfor %}
        
    """, projects=Project.query.all())

@app.route('/add_project', methods=['POST'])
def add_project():
    if request.method == 'POST':
        project_name = request.form['projectName']
        project_description = request.form['projectDescription']
        project_url = request.form['projectUrl']

        # 检查是否已存在，或者直接尝试插入并处理 IntegrityError
        existing_project = Project.query.filter_by(projectName=project_name).first()
        if existing_project:
            print(f"项目 '{project_name}' 已存在，跳过插入。")
            # 可以返回一个错误消息或重定向到带有消息的页面
            return redirect(url_for('index', message='Project already exists!'))

        new_project = Project(
            projectName=project_name,
            projectDescription=project_description,
            projectUrl=project_url
        )
        db.session.add(new_project)
        try:
            db.session.commit()
            print(f"项目 '{project_name}' 插入成功。")
            # 成功后重定向到GET请求的页面
            return redirect(url_for('index', message='Project added successfully!'))
        except IntegrityError:
            db.session.rollback()
            print(f"项目 '{project_name}' 插入失败，可能存在并发冲突。")
            return redirect(url_for('index', message='Failed to add project due to conflict!'))
    # 对于GET请求到/add_project，重定向回首页或显示错误
    return redirect(url_for('index'))

# if __name__ == '__main__':
#     app.run(debug=True)

在这个例子中，当用户提交表单到 /add_project（POST请求）后，数据被处理并存入数据库，然后用户被重定向到 / 页面（GET请求）。这样，即使刷新 / 页面，也不会重复提交表单。

关于 app.app_context() 的正确理解

原始问题中提到“如果将其取出 app.app_context() 就会报错”。这是因为 app.app_context() 提供了一个上下文环境，使得 Flask 扩展（如 Flask-SQLAlchemy）能够正常工作。

app.app_context()： 用于在 Flask 应用之外（例如，在独立的脚本中、命令行工具中或后台任务中）访问应用程序的配置、数据库连接等资源。它模拟了一个应用环境，允许你执行需要访问 current_app 或其扩展的操作。
request_context()： 在处理 Web 请求时，Flask 会自动创建一个请求上下文。在这个上下文中，你可以访问 request、session 等对象，并且 Flask-SQLAlchemy 等扩展也会自动使用这个上下文来管理数据库会话。

因此，在视图函数内部，你通常不需要显式地使用 with app.app_context():，因为请求上下文已经为你提供了必要的环境。但在启动应用时进行数据初始化，或者运行一个独立的管理脚本时，app.app_context() 是必不可少的。

总结与最佳实践

防止 Flask-SQLAlchemy 中数据重复插入的最佳策略通常是结合使用上述两种方法：

数据库层面的唯一性约束是基础： 它是防止数据重复的最强防线，即使应用层逻辑有漏洞，数据库也能保证数据的完整性。
Web 流程中的 POST-Redirect-GET 模式是用户体验的关键： 它避免了用户因无意操作（如刷新、回退）导致重复提交，提升了用户体验。
在应用层进行数据存在性检查（Select-Before-Insert）或错误捕获： 可以在插入前进行检查，提供更友好的用户反馈，或者通过捕获 IntegrityError 来处理并发冲突，确保事务的正确性。
正确使用 app.app_context()： 确保在非请求上下文中操作数据库时，始终显式地进入应用上下文。

通过综合运用这些技术，你可以构建出健壮、可靠的 Flask 应用程序，有效管理数据库中的数据完整性。

PythonSelenium自动化教程_浏览器操作与表单提交实战

Django 开发中何时需要保持开发服务器运行？

Django 开发中是否需要持续运行开发服务器才能执行管理命令？

Python怎么切换成中文界面_Python环境切换中文界面步骤

PythonWebSocket教程_实时通信基础实践