SQLAlchemy IntegrityError：唯一约束失效且数据已入库问题求助

问题描述 #

开发用户注册功能，创建了User模型与表单类接收注册数据。提交表单时弹出错误页面，错误信息显示唯一约束冲突，但对应的注册数据已成功存入数据库。

错误信息 #

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: user.email_adress
[SQL: INSERT INTO user (username, email_adress, password_hash, budget) VALUES (?, ?, ?, ?)]
[parameters: ('erickkpruneda', 'prunefeli@gmail.com', 'Guero432', 1000)]

相关代码 #

路由代码 #

from market import app, db
from flask import render_template, redirect, url_for
from market.models import Item, User
from market.forms import RegisterForms

@app.route('/')   # decorator
def home_page():
    return render_template('home.html')

@app.route('/market')
def market():
    items = Item.query.all()
    return render_template('market.html', items = items)  

@app.route('/register', methods = ['GET', 'POST'])
def registerPage():
    form = RegisterForms()
   
    if form.validate_on_submit():
      
        user_to_create = User(username = form.username.data, email_adress = form.email.data, password_hash = form.password_1.data)
        
        db.session.add(user_to_create)
        db.session.commit()
        
      
        return redirect(url_for('market'))
    return render_template('register.html', form = form)

模型代码 #

from market import db

class User(db.Model):
    id = db.Column(db.Integer(), primary_key = True)
    username = db.Column(db.String(length = 30), nullable = False, unique = True)  
    email_adress = db.Column(db.String(length = 50), nullable = False, unique = True)
    password_hash = db.Column(db.String(length = 60), nullable = False)
    budget = db.Column(db.Integer(), nullable = False, default = 1000)
    items = db.relationship('Item', backref = 'owned_user', lazy = True) 

    def __repr__(self) -> str:
        return f'User {self.username}'

表单代码 #

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField

class RegisterForms(FlaskForm):
 
    username = StringField(label= 'Username: ')
    email = StringField(label= 'Email address: ')
    password_1 = PasswordField(label= 'Password:')
    password_2 = PasswordField(label= 'Confirm password: ')
    submit = SubmitField(label= 'Create account')

原因分析 #

1. 重复数据触发唯一约束：数据库user.email_adress字段设置了unique=True，当提交的邮箱已存在时，db.session.commit()会触发SQLite的唯一性约束错误。你看到数据已存入，大概率是该邮箱记录在之前已成功写入（比如测试提交），本次属于重复提交。
2. 表单缺少唯一性校验：表单类未添加用户名、邮箱的唯一性验证规则，导致重复数据能直接提交到后端，直到数据库层面才抛出错误。
3. 异常未处理导致事务残留：commit()抛出异常时，代码未捕获并回滚会话，可能出现数据已写入的错觉（SQLite默认原子提交，但异常场景下可能存在操作残留）。

解决方案 #

1. 给表单添加唯一性验证 #

修改表单代码，添加前端/后端双重校验：

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length, EqualTo, ValidationError
from market.models import User

class RegisterForms(FlaskForm):
    username = StringField(label='Username: ', validators=[
        DataRequired(),
        Length(min=3, max=30)
    ])
    email = StringField(label='Email address: ', validators=[
        DataRequired(),
        Email()
    ])
    password_1 = PasswordField(label='Password:', validators=[
        DataRequired(),
        Length(min=6)
    ])
    password_2 = PasswordField(label='Confirm password: ', validators=[
        DataRequired(),
        EqualTo('password_1', message='Passwords must match!')
    ])
    submit = SubmitField(label='Create account')

    def validate_username(self, username):
        user = User.query.filter_by(username=username.data).first()
        if user:
            raise ValidationError('Username already exists! Please choose a different one.')
    
    def validate_email(self, email):
        user = User.query.filter_by(email_adress=email.data).first()
        if user:
            raise ValidationError('Email already exists! Please choose a different one.')

2. 后端添加异常捕获与事务回滚 #

修改注册路由，捕获异常并给用户反馈：

from market import app, db
from flask import render_template, redirect, url_for, flash
from market.models import Item, User
from market.forms import RegisterForms
from sqlalchemy.exc import IntegrityError

# ... 其他路由代码 ...

@app.route('/register', methods = ['GET', 'POST'])
def registerPage():
    form = RegisterForms()
   
    if form.validate_on_submit():
        try:
            user_to_create = User(username = form.username.data, 
                                 email_adress = form.email.data, 
                                 password_hash = form.password_1.data)
            
            db.session.add(user_to_create)
            db.session.commit()
            flash('Account created successfully!', category='success')
            return redirect(url_for('market'))
        except IntegrityError:
            db.session.rollback()
            flash('Username or Email already exists!', category='danger')
    return render_template('register.html', form = form)

3. 修复密码存储安全问题 #

当前直接存储明文密码违反安全规范，修改User模型添加哈希处理：

from market import db
from werkzeug.security import generate_password_hash, check_password_hash

class User(db.Model):
    # ... 其他字段 ...
    password_hash = db.Column(db.String(length=60), nullable=False)

    @property
    def password(self):
        raise AttributeError('Password is not a readable attribute!')
    
    @password.setter
    def password(self, password):
        self.password_hash = generate_password_hash(password)
    
    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

# 同时修改路由中的用户创建逻辑：
user_to_create = User(username = form.username.data, 
                     email_adress = form.email.data, 
                     password = form.password_1.data)

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