当前位置：首页 > 文章列表 > 文章 > python教程 > FlaskSQLAlchemy防止重复插入技巧

FlaskSQLAlchemy防止重复插入技巧

2025-12-07 16:33:42 0浏览收藏

推广推荐

支持 PC / 移动端，安全直达

小伙伴们对文章编程感兴趣吗？是否正在学习相关知识点？如果是，那么本文《Flask SQLAlchemy 防止数据重复插入方法》，就很适合你，本篇文章讲解的知识点主要包括。在之后的文章中也会多多分享相关知识点，希望对大家的知识积累有所帮助！

Flask与SQLAlchemy：有效防止数据重复插入的策略

本教程旨在解决在Flask应用中使用SQLAlchemy时，数据重复插入的常见问题。文章将详细介绍两种核心策略：首先，通过数据库层面设置唯一性约束，并结合查询判断或异常处理来确保数据完整性；其次，在Web开发中采用POST-Redirect-GET模式，有效防止因页面刷新导致的重复提交。通过这些方法，开发者可以确保数据的唯一性和一致性，同时优化用户体验。

深入理解数据重复插入问题

在Flask应用中，当使用SQLAlchemy将Python列表（如experience、project）中的数据写入数据库时，经常会遇到数据重复插入的问题。这通常表现为每次应用加载、页面刷新或特定函数被调用时，相同的数据被反复写入数据库，导致数据库中出现大量重复记录。

导致此问题的原因可能包括：

初始化脚本反复执行： 如果数据插入逻辑放在了应用启动时（例如，在__init__.py或某个初始化函数中），那么每次应用重启或在开发模式下热重载时，该逻辑都会被执行。
Web请求处理不当： 如果数据插入逻辑位于一个处理GET请求的视图函数中，或者在POST请求处理后没有进行适当的重定向，用户刷新页面就可能导致数据再次提交。
缺乏数据完整性检查： 在数据插入前没有检查记录是否已存在，或数据库层面没有设置唯一性约束。

解决此问题的关键在于从数据库层面和应用逻辑层面同时进行优化。

策略一：利用数据库唯一性约束确保数据完整性

数据库的唯一性约束是防止数据重复最根本、最有效的方法。通过在模型定义中指定字段的唯一性，数据库会在插入重复数据时抛出错误，从而阻止重复记录的产生。

1. 单字段唯一性约束

对于某个字段的值必须是唯一的场景，可以直接在模型字段定义中添加unique=True参数。

示例代码：

from flask_sqlalchemy import SQLAlchemy
from flask import Flask

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydatabase.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = 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'<Project {self.projectName}>'

# 在应用上下文中创建表
with app.app_context():
    db.create_all()

2. 复合唯一性约束

当多个字段的组合值必须是唯一时（例如，公司名称、角色和持续时间的组合），可以使用UniqueConstraint。

示例代码：

from sqlalchemy import UniqueConstraint

class 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'<Experience {self.companyName} - {self.companyRole}>'

# 在应用上下文中创建表
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 IntegrityError

with 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_string
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydatabase.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = 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'<Item {self.name}>'

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('''
        <h1>Add New Item</h1>
        <form method="POST" action="{{ url_for('add_item') }}">
            <label for="item_name">Item Name:</label><br>
            &lt;input type=&quot;text&quot; id=&quot;item_name&quot; name=&quot;item_name&quot; required&gt;<br><br>
            <label for="item_description">Description:</label><br>
            &lt;textarea id=&quot;item_description&quot; name=&quot;item_description&quot;&gt;&lt;/textarea&gt;<br><br>
            &lt;input type=&quot;submit&quot; value=&quot;Add Item&quot;&gt;
        </form>
        <h2>Existing Items</h2>
        <ul>
            {% for item in items %}
            <li>{{ item.name }} - {{ item.description }}</li>
            {% endfor %}
        </ul>
    ''', items=items)

@app.route('/success')
def success_page():
    item_name = request.args.get('item_name', 'Unknown')
    return render_template_string(f'''
        <h1>Success!</h1>
        <p>Item "{item_name}" has been added successfully.</p>
        <p><a href="{{ url_for('add_item') }}">Add another item</a></p>
    ''')

@app.route('/error')
def error_page():
    message = request.args.get('message', 'An error occurred.')
    return render_template_string(f'''
        <h1>Error!</h1>
        <p>{message}</p>
        <p><a href="{{ url_for('add_item') }}">Try again</a></p>
    ''')

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