本教程旨在解决在flask应用中使用sqlalchemy时，数据重复插入的常见问题。文章将详细介绍两种核心策略：首先，通过数据库层面设置唯一性约束，并结合查询判断或异常处理来确保数据完整性；其次，在web开发中采用post-redirect-get模式，有效防止因页面刷新导致的重复提交。通过这些方法，开发者可以确保数据的唯一性和一致性，同时优化用户体验。
深入理解数据重复插入问题
在Flask应用中，当使用SQLAlchemy将Python列表（如experience、project）中的数据写入数据库时，经常会遇到数据重复插入的问题。这通常表现为每次应用加载、页面刷新或特定函数被调用时，相同的数据被反复写入数据库，导致数据库中出现大量重复记录。
导致此问题的原因可能包括：
初始化脚本反复执行： 如果数据插入逻辑放在了应用启动时（例如，在__init__.py或某个初始化函数中），那么每次应用重启或在开发模式下热重载时，该逻辑都会被执行。Web请求处理不当： 如果数据插入逻辑位于一个处理GET请求的视图函数中，或者在POST请求处理后没有进行适当的重定向，用户刷新页面就可能导致数据再次提交。缺乏数据完整性检查： 在数据插入前没有检查记录是否已存在，或数据库层面没有设置唯一性约束。
解决此问题的关键在于从数据库层面和应用逻辑层面同时进行优化。
策略一：利用数据库唯一性约束确保数据完整性
数据库的唯一性约束是防止数据重复最根本、最有效的方法。通过在模型定义中指定字段的唯一性，数据库会在插入重复数据时抛出错误，从而阻止重复记录的产生。
1. 单字段唯一性约束
对于某个字段的值必须是唯一的场景，可以直接在模型字段定义中添加unique=True参数。
示例代码：
from flask_sqlalchemy import SQLAlchemyfrom flask import Flaskapp = Flask(__name__)app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydatabase.db'app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = Falsedb = SQLAlchemy(app)class Project(db.Model): id = db.Column(db.Integer, primary_key=True) projectName = db.Column(db.String(100), unique=True, nullable=False) # 项目名称必须唯一 projectDescription = db.Column(db.Text) projectUrl = db.Column(db.String(200)) def __repr__(self): return f''# 在应用上下文中创建表with app.app_context(): db.create_all()
2. 复合唯一性约束
当多个字段的组合值必须是唯一时（例如，公司名称、角色和持续时间的组合），可以使用UniqueConstraint。
示例代码：
from sqlalchemy import UniqueConstraintclass Experience(db.Model): id = db.Column(db.Integer, primary_key=True) companyName = db.Column(db.String(100), nullable=False) companyRole = db.Column(db.String(100), nullable=False) companyDuration = db.Column(db.String(50), nullable=False) companyLocation = db.Column(db.String(100)) # ... 其他字段 __table_args__ = ( UniqueConstraint('companyName', 'companyRole', 'companyDuration', name='_company_role_duration_uc'), ) def __repr__(self): return f''# 在应用上下文中创建表with app.app_context(): db.create_all()
3. 处理重复数据插入逻辑
设置了唯一性约束后，尝试插入重复数据会引发IntegrityError。我们可以通过两种主要方式处理：
方法A: 插入前检查是否存在（Get-or-Create模式）
在插入新数据之前，先查询数据库中是否已存在具有相同唯一标识的数据。如果存在，则跳过插入；如果不存在，则进行插入。
示例代码：
from sqlalchemy.exc import IntegrityError# 假设 projectData 和 experience_data 是你的数据列表projectData = [ {"projectName": "Project Alpha", "projectDescription": "Desc A", "projectUrl": "url_a"}, {"projectName": "Project Beta", "projectDescription": "Desc B", "projectUrl": "url_b"}, {"projectName": "Project Alpha", "projectDescription": "Desc A updated", "projectUrl": "url_a_new"}, # 重复项目]experience_data = [ {"companyName": "Comp X", "companyRole": "Dev", "companyDuration": "1 year", "companyLocation": "NY"}, {"companyName": "Comp Y", "companyRole": "QA", "companyDuration": "2 years", "companyLocation": "LA"}, {"companyName": "Comp X", "companyRole": "Dev", "companyDuration": "1 year", "companyLocation": "NY"}, # 重复经验]with app.app_context(): # 处理项目数据 for project_item in projectData: # 检查项目是否已存在 existing_project = Project.query.filter_by(projectName=project_item["projectName"]).first() if not existing_project: project_entry = Project( projectName=project_item["projectName"], projectDescription=project_item["projectDescription"], projectUrl=project_item["projectUrl"], ) db.session.add(project_entry) print(f"Added project: {project_item['projectName']}") else: print(f"Project '{project_item['projectName']}' already exists. Skipping.") # 处理经验数据 for data_item in experience_data: # 检查经验是否已存在 (使用复合唯一性字段进行查询) existing_experience = Experience.query.filter_by( companyName=data_item["companyName"], companyRole=data_item["companyRole"], companyDuration=data_item["companyDuration"] ).first() if not existing_experience: experience_entry = Experience( companyName=data_item["companyName"], companyDescription=data_item.get("companyDescription"), # 使用.get处理可能不存在的字段 companyUrl=data_item.get("companyUrl"), companyRole=data_item["companyRole"], companyDuration=data_item["companyDuration"], companyLocation=data_item["companyLocation"], companyResponsibilities=data_item.get("companyResponsibilities"), # projects=data_item.get("projects") # 如果projects是关系字段，需要额外处理 ) db.session.add(experience_entry) print(f"Added experience: {data_item['companyName']} - {data_item['companyRole']}") else: print(f"Experience for '{data_item['companyName']} - {data_item['companyRole']}' already exists. Skipping.") try: db.session.commit() # 统一提交所有新增数据 print("All new data committed successfully.") except IntegrityError: db.session.rollback() print("An IntegrityError occurred during commit. Rolling back.") except Exception as e: db.session.rollback() print(f"An unexpected error occurred: {e}. Rolling back.")
方法B: 捕获IntegrityError
允许数据库抛出错误，然后捕获并处理它。这种方法适用于你希望在尝试插入后知道是否成功，或者数据库本身提供ON CONFLICT等更高级的语法时。
示例代码：
from sqlalchemy.exc import IntegrityErrorwith app.app_context(): for project_item in projectData: project_entry = Project( projectName=project_item["projectName"], projectDescription=project_item["projectDescription"], projectUrl=project_item["projectUrl"], ) db.session.add(project_entry) try: db.session.commit() # 尝试提交单个项目 print(f"Added project: {project_item['projectName']}") except IntegrityError: db.session.rollback() # 回滚失败的事务 print(f"Project '{project_item['projectName']}' already exists (IntegrityError).") except Exception as e: db.session.rollback() print(f"An unexpected error occurred for project '{project_item['projectName']}': {e}") # 同样可以应用于 experience_data for data_item in experience_data: experience_entry = Experience( companyName=data_item["companyName"], companyRole=data_item["companyRole"], companyDuration=data_item["companyDuration"], companyLocation=data_item["companyLocation"], ) db.session.add(experience_entry) try: db.session.commit() print(f"Added experience: {data_item['companyName']} - {data_item['companyRole']}") except IntegrityError: db.session.rollback() print(f"Experience for '{data_item['companyName']} - {data_item['companyRole']}' already exists (IntegrityError).") except Exception as e: db.session.rollback() print(f"An unexpected error occurred for experience '{data_item['companyName']}': {e}")
注意事项：
在捕获IntegrityError时，务必调用db.session.rollback()来清除当前会话中处于错误状态的事务，否则后续操作可能会失败。如果选择逐个提交，性能可能不如批量提交。如果需要批量处理，且希望在遇到重复时跳过而非终止整个批次，则“插入前检查”模式更合适。
策略二：Web请求中的POST-Redirect-GET模式
如果数据重复插入问题发生在Web请求中（例如用户提交表单后刷新页面），那么采用POST-Redirect-GET (PRG)模式是行业标准实践。
1. PRG模式原理
当用户通过POST请求提交数据到服务器后，服务器处理完数据（如写入数据库）不直接渲染页面，而是发送一个HTTP重定向响应（302 Found）到另一个GET请求的URL。浏览器接收到重定向后，会向新的GET URL发起请求，显示最终结果页面。
这样做的优点是：
防止重复提交： 用户刷新的是GET请求的页面，而不是POST请求，从而避免了重复提交数据。改善用户体验： 页面URL反映了当前的状态，用户可以安全地收藏或分享。避免浏览器警告： 防止浏览器在刷新POST页面时提示“是否重新提交表单数据”。
2. Flask中的PRG实现
示例代码：
from flask import Flask, request, redirect, url_for, render_template_stringfrom flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydatabase.db'app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = Falsedb = SQLAlchemy(app)class Item(db.Model): id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(100), unique=True, nullable=False) description = db.Column(db.Text) def __repr__(self): return f''with app.app_context(): db.create_all()@app.route('/', methods=['GET', 'POST'])def add_item(): if request.method == 'POST': item_name = request.form.get('item_name') item_description = request.form.get('item_description') if item_name: with app.app_context(): existing_item = Item.query.filter_by(name=item_name).first() if not existing_item: new_item = Item(name=item_name, description=item_description) db.session.add(new_item) try: db.session.commit() print(f"Added item: {item_name}") # 成功后重定向到成功页面或列表页面 return redirect(url_for('success_page', item_name=item_name)) except IntegrityError: db.session.rollback() print(f"Item '{item_name}' already exists.") # 重定向到错误页面或带错误信息的表单页 return redirect(url_for('error_page', message=f"Item '{item_name}' already exists.")) else: print(f"Item '{item_name}' already exists (pre-check).") return redirect(url_for('error_page', message=f"Item '{item_name}' already exists.")) # 如果没有item_name，也重定向避免刷新 return redirect(url_for('add_item')) # 或者重定向到带错误信息的页面 # GET 请求显示表单 items = Item.query.all() return render_template_string(''' Add New Item
Existing Items
- {% for item in items %}
- {{ item.name }} - {{ item.description }} {% endfor %}
Success!
Item "{item_name}" has been added successfully.
''')@app.route('/error')def error_page(): message = request.args.get('message', 'An error occurred.') return render_template_string(f'''Error!
{message}
''')if __name__ == '__main__': app.run(debug=True)总结与建议
防止Flask与SQLAlchemy中的数据重复插入是一个多方面的问题，需要结合具体场景采取合适的策略：
数据库层面是基础：
以上就是Flask与SQLAlchemy：有效防止数据重复插入的策略的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1378683.html
微信扫一扫 
支付宝扫一扫 
