本文旨在探讨并解决在使用 flask 和 sqlalchemy 进行数据持久化时，由页面刷新或脚本重复执行导致的数据库数据重复插入问题。我们将深入分析两种核心策略：通过数据库层面的唯一性约束来阻止重复数据，以及利用 web 开发中的 post-redirect-get 模式来避免客户端意外的重复提交，同时也会解释 flask 应用上下文的正确使用。
在使用 Flask 结合 SQLAlchemy 管理数据库时，开发者常会遇到一个常见但令人困扰的问题：数据在不经意间被重复插入数据库。这通常发生在以下场景：
原始代码示例展示了在 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()接下来，我们将探讨两种主要的解决方案来有效防止此类重复插入。
最直接且可靠的方法是在数据库层面强制执行数据的唯一性。这意味着即使应用程序尝试插入重复数据，数据库也会拒绝该操作，并抛出错误。
在 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'<Project {self.projectName}>'示例：为 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'<Experience {self.companyName} - {self.companyRole}>'当尝试插入违反唯一性约束的数据时，数据库会抛出 IntegrityError。在应用程序中，需要捕获并处理这个错误。
方法一：先查询后插入 (Select-Before-Insert)
在插入数据前，先查询数据库中是否已存在具有相同唯一标识的数据。
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)在 Web 应用中，当用户提交表单（通常是 POST 请求）后，如果直接渲染页面或返回数据，用户刷新浏览器可能会导致浏览器重新发送 POST 请求，从而引发重复提交。POST-Redirect-GET (PRG) 模式是解决此问题的标准实践。
通过这种方式，即使用户刷新浏览器，也只会刷新最后的 GET 请求页面，而不会重新发送 POST 请求。
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'<Project {self.projectName}>'
# 在应用上下文外创建表 (通常在独立脚本或第一次运行时)
with app.app_context():
db.create_all()
@app.route('/', methods=['GET'])
def index():
# 显示一个简单的表单
return render_template_string("""
<h1>添加新项目</h1>
<form method="POST" action="/add_project">
项目名称: <input type="text" name="projectName"><br>
项目描述: <textarea name="projectDescription"></textarea><br>
项目URL: <input type="text" name="projectUrl"><br>
<input type="submit" value="提交">
</form>
<h2>现有项目</h2>
<ul>
{% for project in projects %}
<li>{{ project.projectName }} - {{ project.projectDescription }}</li>
{% endfor %}
</ul>
""", 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() 提供了一个上下文环境，使得 Flask 扩展（如 Flask-SQLAlchemy）能够正常工作。
因此，在视图函数内部，你通常不需要显式地使用 with app.app_context():，因为请求上下文已经为你提供了必要的环境。但在启动应用时进行数据初始化，或者运行一个独立的管理脚本时，app.app_context() 是必不可少的。
防止 Flask-SQLAlchemy 中数据重复插入的最佳策略通常是结合使用上述两种方法：
通过综合运用这些技术，你可以构建出健壮、可靠的 Flask 应用程序，有效管理数据库中的数据完整性。
以上就是Flask-SQLAlchemy 数据重复插入问题及解决方案的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告
收藏
收藏
收藏
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
714
