Flask注册接口触发sqlalchemy.exc.IntegrityError问题排查与解决

在使用Flask+SQLAlchemy+PostgreSQL开发用户注册功能时，调用/registeruser接口同时传入email和mobile_number参数，触发sqlalchemy.exc.IntegrityError: (psycopg2.errors.NotNullViolation)错误，预期插入的邮箱和手机号字段未正确写入，最终发现是原代码逻辑存在对象覆盖问题，修正逻辑后问题解决。

原模型代码（models.py） #

class UsersT(db.Model):
    __tablename__ = "userst"
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(100), unique=True, nullable=True)
    firstname = db.Column(db.String(100), nullable=True)
    lastname = db.Column(db.String(100), nullable=True)
    city = db.Column(db.String(50), nullable=True)
    mobile_number = db.Column(db.String(20), nullable=True, unique=True)
    created = db.Column(db.DateTime(timezone=True), default=func.now())
    createdby = db.Column(db.String(100), nullable=True)
    modified = db.Column(db.DateTime(timezone=True), default=func.now(), onupdate=func.now(), nullable=True)
    modifiedby = db.Column(db.String(100), nullable=True)
    status = db.Column(db.String(20), nullable=True)
    statusreason = db.Column(db.String(100), nullable=True)

原接口代码（app.py） #

class RegisterUser(Resource):
    def post(self):
        parser = reqparse.RequestParser()
        parser.add_argument('email', type=str, required=False)
        parser.add_argument('mobile_number', type=str, required=False)
        args = parser.parse_args()

        email = args.get('email')
        mobile_number = args.get('mobile_number')

        print("Received email:", email)
        print("Received mobile_number:", mobile_number)

        new_user = None
        
        if not email and not mobile_number:
            return jsonify({'error': 'email or mobile number is required'}), 400
        
        if email and mobile_number:
            new_user = UsersT(mobile_number=mobile_number)
            entered_value = mobile_number

        if email:
            existing_user = UsersT.query.filter_by(email=email).first()
            if existing_user:
                return jsonify({'error': 'email already exists. try logging in'}), 400
            else: 
                new_user = UsersT(email=email)
                entered_value = email

        elif mobile_number:
            existing_user = UsersT.query.filter_by(mobile_number=mobile_number).first()
            if existing_user:
                return jsonify({'error': 'mobile number already exists. try logging in'}), 400
            else:
                new_user = UsersT(mobile_number=mobile_number)
                entered_value = mobile_number

        if new_user:
            db.session.add(new_user)
            db.session.commit()
            return "User registered successfully", jsonify({'user_id': new_user.id, 'entered_value': entered_value}), 200

        else:
            return jsonify({'error': 'No valid user data provided'}), 400
        
        user_id = new_user.id

报错信息 #

sqlalchemy.exc.IntegrityError: (psycopg2.errors.NotNullViolation)
null value in column "mobile_number" of relation "userst" violates
not-null constraint DETAIL: Failing row contains (25, johndoe@gm.com,
null, null, null, null, null, null, null, null, null, null, 2024-05-10
13:35:36.864429+05:30, null, 2024-05-10 13:35:36.864429+05:30, null,
null, null). [SQL: INSERT INTO userst (email, city, mobile_number,
created, createdby, modified, modifiedby, status, statusreason) VALUES
(%(email)s, %(city)s, %(mobile_number)s, now(), %(createdby)s, now(),
%(modifiedby)s, %(status)s, %(statusreason)s) RETURNING userst.id,
userst.created, userst.modified] [parameters: {'email':
'johndoe@gm.com', 'city': None, 'mobile_number': None, 'createdby':
None, 'modifiedby': None, 'status': None, 'statusreason': None}]

问题根源 #

原代码逻辑存在对象覆盖问题：当同时传入email和mobile_number时，先执行if email and mobile_number分支创建了包含手机号的new_user，但后续的if email分支会被触发，重新创建仅包含邮箱的new_user，直接覆盖了之前的对象。最终插入数据库时，mobile_number字段为null，违反了数据库的非空约束（注：模型中nullable=True但数据库实际约束可能为非空）。

修正后的接口代码 #

email = args.get('email')
mobile_number = args.get('mobile_number')

print("Received email:", email)
print("Received mobile_number:", mobile_number)

new_user = None

if not email and not mobile_number:
    return jsonify({'error': 'email or mobile number is required'}), 400

if mobile_number:
    existing_user = UsersT.query.filter_by(mobile_number=mobile_number).first()
else:
    existing_user = UsersT.query.filter_by(email=email).first()

if existing_user:
    return jsonify({'error': 'User already exists'}), 400

if mobile_number:
    new_user = UsersT(mobile_number=mobile_number)
    entered_value = mobile_number
else:
    new_user = UsersT(email=email)
    entered_value = email

if new_user:
    db.session.add(new_user)
    db.session.commit()
    return {'user_id': new_user.id, 'entered_value': entered_value}, 200

else:
    return jsonify({'error': 'No valid user data provided'}), 400

修复说明 #

修正后的代码调整了逻辑顺序：

1. 先校验必填参数是否存在；
2. 优先根据传入的手机号/邮箱检查用户是否已存在；
3. 最后根据优先级创建用户对象，避免了原代码中对象被重复创建覆盖的问题，确保传入的参数能正确映射到数据库字段。

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