FlaskSQLAlchemy防重复插入技巧分享

最近发现不少小伙伴都对文章很感兴趣，所以今天继续给大家介绍文章相关的知识，本文《Flask SQLAlchemy防重复插入方法与技巧》主要内容涉及到等等知识点，希望能帮到你！当然如果阅读本文时存在不同想法，可以在评论中表达，但是请勿使用过激的措辞~

本文旨在探讨在Flask应用中使用SQLAlchemy将列表数据插入数据库时，如何有效避免数据重复插入的问题。我们将深入分析导致重复的常见原因，并提供两种核心策略：一是利用数据库的唯一性约束进行数据校验与插入，二是采用Web开发中的POST-Redirect-GET模式来防止用户意外刷新导致的重复提交，确保数据持久化过程的健壮性和准确性。

在Flask应用中，利用SQLAlchemy进行数据持久化是常见的操作。然而，当我们需要从Python列表（如字典列表）批量插入数据时，如果不采取适当的预防措施，很容易导致数据重复插入的问题。本教程将详细介绍如何避免这种常见陷阱，确保数据的完整性。

1. 理解数据重复插入的根本原因

在提供的场景中，数据重复插入的主要原因通常在于：

1. 缺乏唯一性检查： 在每次执行插入操作时，代码没有检查即将插入的数据是否已经存在于数据库中。
2. 不当的Web请求处理： 如果数据插入逻辑是在一个GET请求处理函数中，并且该页面被用户多次访问或刷新，就会导致数据被反复插入。
3. 应用程序上下文的误解： app.app_context() 确保了在应用上下文之外也能执行数据库操作，但这与防止重复插入是两个独立的问题。它解决了“应用在上下文之外运行”的错误，但不会阻止重复数据。

2. 策略一：利用数据库唯一性约束防止重复

数据库的唯一性约束是防止数据重复最根本和最有效的方法。它从数据库层面保证了特定字段或字段组合的唯一性。

2.1 声明唯一性约束

在SQLAlchemy模型中，可以通过两种方式声明唯一性约束：

• 单个字段唯一： 为模型中的某个字段添加 unique=True 参数。
• 多个字段组合唯一： 使用 UniqueConstraint 来定义一个或多个字段的组合必须是唯一的。

示例模型定义：

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(120), unique=True, nullable=False) # 项目名称唯一
    projectDescription = db.Column(db.Text, nullable=True)
    projectUrl = db.Column(db.String(255), nullable=True)

    def __repr__(self):
        return f'<Project {self.projectName}>'

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(255), nullable=True)
    companyRole = db.Column(db.String(120), nullable=False)
    companyDuration = db.Column(db.String(120), nullable=True)
    companyLocation = db.Column(db.String(120), nullable=True)
    companyResponsibilities = db.Column(db.Text, nullable=True)
    # 假设公司名称、角色和持续时间组合是唯一的
    __table_args__ = (UniqueConstraint('companyName', 'companyRole', 'companyDuration', name='_company_role_duration_uc'),)

    def __repr__(self):
        return f'<Experience {self.companyName} - {self.companyRole}>'

2.2 插入前检查现有数据

声明唯一性约束后，当尝试插入重复数据时，数据库会抛出错误（如 IntegrityError）。为了更优雅地处理这种情况，我们可以在插入之前查询数据库，检查数据是否已存在。

数据插入逻辑优化：

from flask import Flask
from your_models import db, Project, Experience # 假设模型定义在 your_models.py 中

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

# 示例数据
project_data = [
    {"projectName": "项目A", "projectDescription": "描述A", "projectUrl": "urlA"},
    {"projectName": "项目B", "projectDescription": "描述B", "projectUrl": "urlB"},
    {"projectName": "项目A", "projectDescription": "描述AA", "projectUrl": "urlAA"}, # 重复的项目名称
]

experience_data = [
    {"companyName": "公司X", "companyDescription": "描述X", "companyUrl": "urlX", "companyRole": "工程师", "companyDuration": "1年", "companyLocation": "北京", "companyResponsibilities": "开发"},
    {"companyName": "公司Y", "companyDescription": "描述Y", "companyUrl": "urlY", "companyRole": "设计师", "companyDuration": "2年", "companyLocation": "上海", "companyResponsibilities": "设计"},
    {"companyName": "公司X", "companyDescription": "描述XX", "companyUrl": "urlXX", "companyRole": "工程师", "companyDuration": "1年", "companyLocation": "广州", "companyResponsibilities": "维护"}, # 重复的组合
]

def add_initial_data():
    with app.app_context():
        # 确保数据库表已创建
        db.create_all()

        # 添加项目数据
        for project_item in project_data:
            # 检查项目名称是否已存在
            existing_project = Project.query.filter_by(projectName=project_item["projectName"]).first()
            if existing_project:
                print(f"项目 '{project_item['projectName']}' 已存在，跳过插入。")
            else:
                project_entry = Project(
                    projectName=project_item["projectName"],
                    projectDescription=project_item["projectDescription"],
                    projectUrl=project_item["projectUrl"],
                )
                db.session.add(project_entry)
                print(f"添加项目: {project_item['projectName']}")

        # 添加经验数据
        for exp_item in experience_data:
            # 检查经验数据组合是否已存在
            existing_experience = Experience.query.filter_by(
                companyName=exp_item["companyName"],
                companyRole=exp_item["companyRole"],
                companyDuration=exp_item["companyDuration"]
            ).first()
            if existing_experience:
                print(f"经验 '{exp_item['companyName']} - {exp_item['companyRole']} - {exp_item['companyDuration']}' 已存在，跳过插入。")
            else:
                experience_entry = Experience(
                    companyName=exp_item["companyName"],
                    companyDescription=exp_item["companyDescription"],
                    companyUrl=exp_item["companyUrl"],
                    companyRole=exp_item["companyRole"],
                    companyDuration=exp_item["companyDuration"],
                    companyLocation=exp_item["companyLocation"],
                    companyResponsibilities=exp_item["companyResponsibilities"]
                    # 注意：如果 projects 是关系字段，需要单独处理或在创建关系时建立
                )
                db.session.add(experience_entry)
                print(f"添加经验: {exp_item['companyName']}")

        try:
            db.session.commit()
            print("所有数据提交成功。")
        except Exception as e:
            db.session.rollback()
            print(f"数据提交失败: {e}")

if __name__ == '__main__':
    add_initial_data()
    # 可以在此处运行 Flask 应用
    # app.run(debug=True)

注意事项：

• db.session.add() 只是将对象添加到会话中，真正的数据库操作发生在 db.session.commit()。
• 将所有 add 操作放在循环内部，然后一次性 commit，可以提高性能并确保事务的原子性。
• 如果数据量大，可以考虑使用 db.session.add_all() 批量添加对象。

3. 策略二：通过Web请求处理避免重复提交

如果数据插入是通过Web请求触发的（例如，用户提交表单），那么采用POST-Redirect-GET (PRG) 模式是防止用户刷新页面导致重复提交的有效方法。

3.1 POST-Redirect-GET (PRG) 模式

• POST 请求： 用户提交表单数据时，使用 POST 方法发送请求到服务器。
• 处理数据并重定向： 服务器接收到 POST 请求后，处理数据（例如，将其插入数据库）。处理完成后，不是直接渲染页面，而是向客户端发送一个重定向（HTTP 302 Found）响应，将其引导到一个 GET 请求的URL。
• GET 请求： 客户端收到重定向后，会发送一个新的 GET 请求到指定的URL，服务器再渲染并返回页面。

这样，即使用户刷新了 GET 请求的页面，也只是重新获取了显示页面，而不会再次触发 POST 请求中的数据插入逻辑。

Flask路由示例：

from flask import Flask, request, redirect, url_for, render_template
from your_models import db, Project # 假设 Project 模型已定义

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

# 创建数据库表（如果尚未创建）
with app.app_context():
    db.create_all()

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

        if project_name:
            with app.app_context():
                existing_project = Project.query.filter_by(projectName=project_name).first()
                if existing_project:
                    print(f"项目 '{project_name}' 已存在，不重复添加。")
                    # 可以添加 flash 消息提示用户
                else:
                    new_project = Project(
                        projectName=project_name,
                        projectDescription=project_desc,
                        projectUrl=project_url
                    )
                    db.session.add(new_project)
                    try:
                        db.session.commit()
                        print(f"项目 '{project_name}' 添加成功。")
                    except Exception as e:
                        db.session.rollback()
                        print(f"添加项目失败: {e}")
                        # 可以添加 flash 消息提示错误
        return redirect(url_for('list_projects')) # 重定向到项目列表页
    return render_template('add_project_form.html') # 显示添加项目的表单

@app.route('/projects')
def list_projects():
    with app.app_context():
        projects = Project.query.all()
    return render_template('projects.html', projects=projects)

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

add_project_form.html 示例：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>添加项目</title>
</head>
<body>
    <h1>添加新项目</h1>
    <form method="POST" action="{{ url_for('add_project') }}">
        <label for="projectName">项目名称:</label><br>
        &lt;input type=&quot;text&quot; id=&quot;projectName&quot; name=&quot;projectName&quot; required&gt;<br><br>
        <label for="projectDescription">项目描述:</label><br>
        &lt;textarea id=&quot;projectDescription&quot; name=&quot;projectDescription&quot;&gt;&lt;/textarea&gt;<br><br>
        <label for="projectUrl">项目URL:</label><br>
        &lt;input type=&quot;url&quot; id=&quot;projectUrl&quot; name=&quot;projectUrl&quot;&gt;<br><br>
        &lt;input type=&quot;submit&quot; value=&quot;提交项目&quot;&gt;
    </form>
    <p><a href="{{ url_for('list_projects') }}">查看所有项目</a></p>
</body>
</html>

projects.html 示例：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>项目列表</title>
</head>
<body>
    <h1>项目列表</h1>
    <ul>
        {% for project in projects %}
            <li>{{ project.projectName }} - {{ project.projectDescription }} ({{ project.projectUrl }})</li>
        {% else %}
            <li>暂无项目。</li>
        {% endfor %}
    </ul>
    <p><a href="{{ url_for('add_project') }}">添加新项目</a></p>
</body>
</html>

4. 结合使用与最佳实践

• 双重保障： 数据库唯一性约束是数据完整性的最终保障，而PRG模式则侧重于改善用户体验，防止意外操作。在实际应用中，这两种策略通常会结合使用，形成强大的防御机制。
• 错误处理： 当数据库唯一性约束被触发时，SQLAlchemy会抛出 IntegrityError。在代码中应该捕获并处理这类异常，例如回滚事务并向用户显示友好的错误消息。
• 事务管理： db.session.add() 和 db.session.commit() 应该在适当的事务边界内使用。通常，一个逻辑操作（如添加一个项目或一批项目）应该在一个事务中完成。
• app.app_context() 的理解： app.app_context() 是为了在没有活动请求上下文（例如在脚本或后台任务中）时，也能访问Flask应用配置和扩展（如SQLAlchemy）。在Web请求处理函数中，请求上下文会自动激活，通常不需要手动调用 with app.app_context():。但在独立脚本中初始化数据时，它是必需的。

总结

防止Flask SQLAlchemy中数据重复插入需要多方面考虑。核心策略包括在数据库层面建立唯一性约束，并在插入数据前进行存在性检查；同时，对于通过Web界面提交的数据，应采用POST-Redirect-GET模式来避免重复提交。通过结合运用这些方法，可以有效提高数据管理的健壮性和应用的可靠性。

理论要掌握，实操不能落！以上关于《FlaskSQLAlchemy防重复插入技巧分享》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！