Flask SQLAlchemy 如何实现仅允许一个超级管理员的自定义完整性约束

首先修正你现有代码中的两处基础错误：

1. User表的role_id外键关联对象错误，你当前写的是关联users.id（用户表自身主键），实际应该关联roles.id
2. User表的role关系关联对象错误，应该关联Role模型而非User模型

你可以从以下两种方式中选择符合你场景的实现方案： #

方式一：数据库层面条件唯一索引（推荐） #

这种方式直接在数据库层面加约束，不会被绕过，可靠性最高，支持SQLite、PostgreSQL、MySQL 8.0.13+版本。
你只需要给User模型添加__table_args__配置条件唯一索引即可，该约束会限制当role_id=1（对应superadmin角色）时，users表中只能存在一条符合条件的记录，其他role_id不受影响。

class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True)
    # 其他用户字段...
    # 修正后的外键配置
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id', onupdate='CASCADE'))
    role = db.relationship('Role')

    # 添加条件唯一索引
    __table_args__ = (
        db.UniqueConstraint(
            'role_id',
            name='unique_superadmin',
            sqlite_where='role_id=1',
            mysql_where='role_id=1',
            postgresql_where='role_id=1'
        ),
    )

方式二：ORM层面事件监听（兼容所有数据库） #

如果你的数据库版本不支持条件唯一索引，可以用SQLAlchemy的事件监听器在写入数据前做校验，该方式兼容性强，所有数据库都支持，但仅在通过ORM执行插入/更新操作时生效。

from sqlalchemy import event

# 放在User模型定义外的监听函数
@event.listens_for(User, 'before_insert')
@event.listens_for(User, 'before_update')
def check_unique_superadmin(mapper, connection, target):
    # 动态获取superadmin角色id，避免硬编码写死1更灵活
    superadmin_role = Role.query.filter_by(name='superadmin').first()
    if not superadmin_role:
        return
    if target.role_id == superadmin_role.id:
        exist_superadmin = User.query.filter_by(role_id=superadmin_role.id).first()
        # 插入时已存在/更新时存在的超级管理员不是当前用户，就抛出异常
        if exist_superadmin and exist_superadmin.id != target.id:
            raise ValueError("系统仅允许存在1个超级管理员账号")

完整修正后的参考代码 #

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import event

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"]="sqlite:////tmp/stack.db"
db = SQLAlchemy(app)

class Role(db.Model):
    __tablename__ = "roles"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, unique=True)

    def __repr__(self):
        return self.name

class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True)
    # 其他用户字段...
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id', onupdate='CASCADE'))
    role = db.relationship('Role')

    # 不需要方式一可以注释掉下面的__table_args__
    __table_args__ = (
        db.UniqueConstraint(
            'role_id',
            name='unique_superadmin',
            sqlite_where='role_id=1',
            mysql_where='role_id=1',
            postgresql_where='role_id=1'
        ),
    )

# 不需要方式二可以注释掉下面的监听代码
@event.listens_for(User, 'before_insert')
@event.listens_for(User, 'before_update')
def check_unique_superadmin(mapper, connection, target):
    superadmin_role = Role.query.filter_by(name='superadmin').first()
    if not superadmin_role:
        return
    if target.role_id == superadmin_role.id:
        exist_superadmin = User.query.filter_by(role_id=superadmin_role.id).first()
        if exist_superadmin and exist_superadmin.id != target.id:
            raise ValueError("系统仅允许存在1个超级管理员账号")

内容的提问来源于stack exchange，提问作者A. Josh