在Flask中用SQLAlchemy批量添加列表数据至数据库时重复提交问题

问题背景 #

我有如下数据列表：

experience =[{}, {}, {}] # 包含3个对象的列表
project = [{}...]

已创建对应的Flask-SQLAlchemy模型，尝试通过以下代码将数据添加至数据库：

# 在app_context()函数内添加数据到数据库
with app.app_context():
    # 提交更改

    # 向数据库添加项目数据
    for project in projectData:
        project_entry = Project(
            projectName=project["projectName"],
            projectDescription=project["projectDescription"],
            projectUrl=project["projectUrl"],
        )
        db.session.add(project_entry)

    # 向数据库添加工作经历数据
    for data in experience_data:
        experience_entry = Experience(
            companyName=data["companyName"],
            companyDescription=data["companyDescription"],
            companyUrl=data["companyUrl"],
            companyRole=data["companyRole"],
            companyDuration=data["companyDuration"],
            companyLocation=data["companyLocation"],
            companyResponsibilities=data["companyResponsibilities"],
            projects=data["projects"]
        )
        db.session.add(experience_entry)

    # 添加完所有数据后提交更改
    db.session.commit()

遇到的问题：每次加载时数据都会重复提交，预期仅3行数据但实际已有54行；若移出app.app_context()则会触发应用上下文错误，如何实现仅一次性将数据存储至数据库？

解决方法 #

1. 插入前检查数据是否已存在 #

在添加每条数据前，通过唯一标识字段（如项目名、公司名+职位组合）查询数据库，仅当记录不存在时才插入：

with app.app_context():
    # 处理项目数据
    for project in projectData:
        existing_project = Project.query.filter_by(projectName=project["projectName"]).first()
        if not existing_project:
            project_entry = Project(
                projectName=project["projectName"],
                projectDescription=project["projectDescription"],
                projectUrl=project["projectUrl"],
            )
            db.session.add(project_entry)

    # 处理工作经历数据
    for data in experience_data:
        # 以公司名+职位作为唯一判断条件，可根据实际业务调整
        existing_experience = Experience.query.filter_by(
            companyName=data["companyName"],
            companyRole=data["companyRole"]
        ).first()
        if not existing_experience:
            experience_entry = Experience(
                companyName=data["companyName"],
                companyDescription=data["companyDescription"],
                companyUrl=data["companyUrl"],
                companyRole=data["companyRole"],
                companyDuration=data["companyDuration"],
                companyLocation=data["companyLocation"],
                companyResponsibilities=data["companyResponsibilities"],
                projects=data["projects"]
            )
            db.session.add(experience_entry)

    db.session.commit()

2. 编写独立初始化脚本，仅手动执行一次 #

将数据插入逻辑从应用启动代码中剥离，做成单独的初始化脚本，只运行一次：

创建init_db.py文件：

# 导入你的应用、数据库模型和数据
from your_app import app, db, Project, Experience
from your_data_source import projectData, experience_data

with app.app_context():
    # 先创建数据库表（如果还未创建）
    db.create_all()
    
    # 插入项目数据（带存在性检查）
    for project in projectData:
        if not Project.query.filter_by(projectName=project["projectName"]).first():
            project_entry = Project(**project)
            db.session.add(project_entry)
    
    # 插入工作经历数据（带存在性检查）
    for data in experience_data:
        if not Experience.query.filter_by(companyName=data["companyName"], companyRole=data["companyRole"]).first():
            experience_entry = Experience(**data)
            db.session.add(experience_entry)
    
    db.session.commit()
    print("数据初始化完成，已成功插入非重复数据")

在终端执行一次该脚本：

python init_db.py

之后启动应用时，这段插入逻辑不会重复执行，也就不会产生重复数据。

3. 给数据库添加唯一约束 #

在模型类中为易重复的字段添加唯一约束（单个字段或联合字段），从数据库层面阻止重复插入：

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))

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)
    # 添加公司名+职位的联合唯一约束
    __table_args__ = (
        db.UniqueConstraint('companyName', 'companyRole', name='_company_role_unique'),
    )
    # 其他字段...

插入时捕获数据库的完整性错误，跳过重复数据：

from sqlalchemy.exc import IntegrityError

with app.app_context():
    try:
        # 批量添加项目数据
        for project in projectData:
            project_entry = Project(**project)
            db.session.add(project_entry)
        
        # 批量添加工作经历数据
        for data in experience_data:
            experience_entry = Experience(**data)
            db.session.add(experience_entry)
        
        db.session.commit()
    except IntegrityError:
        db.session.rollback()
        print("检测到重复数据，已回滚操作，未插入重复记录")

内容的提问来源于stack exchange，提问作者dave