Flask-SQLAlchemy如何实现dat_pin与clk_pin全局值唯一约束？

这里有几个可行的方案来实现你需要的全局引脚唯一性约束，确保新添加的dat_pin或clk_pin不会和表中任何已有的引脚值重复，根据你的数据库环境和需求可以选择合适的方式：

方案一：数据库表级CHECK约束（适用于PostgreSQL等支持子查询的数据库） #

如果你的数据库支持在CHECK约束中使用子查询（比如PostgreSQL），可以直接在Sensor模型的__table_args__中添加表级约束，从数据库层面保证唯一性：

class Sensor(db.Model):
    __table_args__ = (
        db.CheckConstraint('dat_pin != clk_pin', name='data_pin_not_eq_clock_pin'),
        # 确保dat_pin未被任何传感器的dat_pin或clk_pin使用
        db.CheckConstraint(
            'NOT EXISTS (SELECT 1 FROM sensor WHERE dat_pin = NEW.dat_pin OR clk_pin = NEW.dat_pin)',
            name='dat_pin_unique_across_all_pins'
        ),
        # 确保clk_pin未被任何传感器的dat_pin或clk_pin使用
        db.CheckConstraint(
            'NOT EXISTS (SELECT 1 FROM sensor WHERE dat_pin = NEW.clk_pin OR clk_pin = NEW.clk_pin)',
            name='clk_pin_unique_across_all_pins'
        ),
    )
    # 其余字段和方法保持不变
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, nullable=False,)
    dat_pin = db.Column(db.Integer, nullable=False,)
    clk_pin = db.Column(db.Integer, nullable=False,)
    zero_val = db.Column(db.Integer, unique=False, nullable=True,)
    cal_factor = db.Column(db.Integer, unique=False, nullable=True,)
    full_val = db.Column(db.Integer, unique=False, nullable=True,)
    alert_val = db.Column(db.Integer, unique=False, nullable=True,)
    values = db.relationship('SensorValues', backref='author', lazy=True)

    def __repr__(self):
        return f"Sensor('{self.name}', '{self.dat_pin}', '{self.clk_pin}', '{self.zero_val}', '{self.cal_factor}', '{self.full_val}', '{self.alert_val}')"

注意：这个方案依赖数据库对表级CHECK约束的支持，SQLite默认不允许在CHECK中使用子查询，所以如果用SQLite的话这个方法不适用。

方案二：SQLAlchemy事件监听（兼容所有数据库） #

如果需要兼容多种数据库（比如SQLite），可以用SQLAlchemy的事件监听机制，在插入或更新前手动验证引脚的唯一性：

from sqlalchemy import event

class Sensor(db.Model):
    __table_args__ = (
        db.CheckConstraint('dat_pin != clk_pin', name='data_pin_not_eq_clock_pin'),
    )
    # 其余字段和方法保持不变
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, nullable=False,)
    dat_pin = db.Column(db.Integer, nullable=False,)
    clk_pin = db.Column(db.Integer, nullable=False,)
    zero_val = db.Column(db.Integer, unique=False, nullable=True,)
    cal_factor = db.Column(db.Integer, unique=False, nullable=True,)
    full_val = db.Column(db.Integer, unique=False, nullable=True,)
    alert_val = db.Column(db.Integer, unique=False, nullable=True,)
    values = db.relationship('SensorValues', backref='author', lazy=True)

    def __repr__(self):
        return f"Sensor('{self.name}', '{self.dat_pin}', '{self.clk_pin}', '{self.zero_val}', '{self.cal_factor}', '{self.full_val}', '{self.alert_val}')"

# 插入前验证
@event.listens_for(Sensor, 'before_insert')
def validate_pins_before_insert(mapper, connection, target):
    # 检查dat_pin是否已被使用
    pin_in_use = connection.query(Sensor).filter(
        (Sensor.dat_pin == target.dat_pin) | (Sensor.clk_pin == target.dat_pin)
    ).first()
    if pin_in_use:
        raise ValueError(f"引脚 {target.dat_pin} 已被其他传感器使用（作为数据引脚或时钟引脚）")
    
    # 检查clk_pin是否已被使用
    pin_in_use = connection.query(Sensor).filter(
        (Sensor.dat_pin == target.clk_pin) | (Sensor.clk_pin == target.clk_pin)
    ).first()
    if pin_in_use:
        raise ValueError(f"引脚 {target.clk_pin} 已被其他传感器使用（作为数据引脚或时钟引脚）")

# 更新前验证（排除当前记录本身）
@event.listens_for(Sensor, 'before_update')
def validate_pins_before_update(mapper, connection, target):
    # 检查dat_pin是否被其他记录使用
    pin_in_use = connection.query(Sensor).filter(
        (Sensor.id != target.id) & 
        ((Sensor.dat_pin == target.dat_pin) | (Sensor.clk_pin == target.dat_pin))
    ).first()
    if pin_in_use:
        raise ValueError(f"引脚 {target.dat_pin} 已被其他传感器使用（作为数据引脚或时钟引脚）")
    
    # 检查clk_pin是否被其他记录使用
    pin_in_use = connection.query(Sensor).filter(
        (Sensor.id != target.id) & 
        ((Sensor.dat_pin == target.clk_pin) | (Sensor.clk_pin == target.clk_pin))
    ).first()
    if pin_in_use:
        raise ValueError(f"引脚 {target.clk_pin} 已被其他传感器使用（作为数据引脚或时钟引脚）")

这个方案的优点是兼容所有数据库，缺点是在高并发场景下可能存在竞态条件（两个请求同时验证通过并插入重复引脚），如果需要绝对的唯一性保障，最好配合数据库层面的约束作为兜底。

方案三：辅助表管理引脚使用（更灵活的扩展方式） #

如果未来可能需要对引脚使用做更复杂的管理（比如记录引脚类型、使用状态），可以创建一个单独的PinUsage表来管理所有被使用的引脚，通过外键和唯一约束来保证唯一性：

class PinUsage(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    pin = db.Column(db.Integer, unique=True, nullable=False)
    sensor_id = db.Column(db.Integer, db.ForeignKey('sensor.id'))
    pin_type = db.Column(db.String(16), nullable=False)  # 标记是dat_pin还是clk_pin
    sensor = db.relationship('Sensor', backref=db.backref('pin_usages', cascade='all, delete-orphan'))

class Sensor(db.Model):
    __table_args__ = (
        db.CheckConstraint('dat_pin != clk_pin', name='data_pin_not_eq_clock_pin'),
    )
    # 其余字段和方法保持不变
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, nullable=False,)
    dat_pin = db.Column(db.Integer, nullable=False,)
    clk_pin = db.Column(db.Integer, nullable=False,)
    zero_val = db.Column(db.Integer, unique=False, nullable=True,)
    cal_factor = db.Column(db.Integer, unique=False, nullable=True,)
    full_val = db.Column(db.Integer, unique=False, nullable=True,)
    alert_val = db.Column(db.Integer, unique=False, nullable=True,)
    values = db.relationship('SensorValues', backref='author', lazy=True)

    def __repr__(self):
        return f"Sensor('{self.name}', '{self.dat_pin}', '{self.clk_pin}', '{self.zero_val}', '{self.cal_factor}', '{self.full_val}', '{self.alert_val}')"

# 插入传感器后自动添加引脚使用记录
@event.listens_for(Sensor, 'after_insert')
def add_pin_usages_after_insert(mapper, connection, target):
    # 添加dat_pin记录
    connection.execute(PinUsage.__table__.insert().values(
        pin=target.dat_pin, sensor_id=target.id, pin_type='dat'
    ))
    # 添加clk_pin记录
    connection.execute(PinUsage.__table__.insert().values(
        pin=target.clk_pin, sensor_id=target.id, pin_type='clk'
    ))

# 更新传感器时处理引脚变化
@event.listens_for(Sensor, 'before_update')
def update_pin_usages_before_update(mapper, connection, target):
    old_sensor = connection.query(Sensor).filter_by(id=target.id).first()
    
    # 处理dat_pin变化
    if old_sensor.dat_pin != target.dat_pin:
        # 删除旧的dat_pin记录
        connection.execute(PinUsage.__table__.delete().filter_by(
            pin=old_sensor.dat_pin, sensor_id=target.id, pin_type='dat'
        ))
        # 检查新引脚是否已被使用
        if connection.query(PinUsage).filter_by(pin=target.dat_pin).first():
            raise ValueError(f"引脚 {target.dat_pin} 已被使用")
        # 添加新的dat_pin记录
        connection.execute(PinUsage.__table__.insert().values(
            pin=target.dat_pin, sensor_id=target.id, pin_type='dat'
        ))
    
    # 处理clk_pin变化
    if old_sensor.clk_pin != target.clk_pin:
        connection.execute(PinUsage.__table__.delete().filter_by(
            pin=old_sensor.clk_pin, sensor_id=target.id, pin_type='clk'
        ))
        if connection.query(PinUsage).filter_by(pin=target.clk_pin).first():
            raise ValueError(f"引脚 {target.clk_pin} 已被使用")
        connection.execute(PinUsage.__table__.insert().values(
            pin=target.clk_pin, sensor_id=target.id, pin_type='clk'
        ))

这个方案通过PinUsage表的pin字段唯一约束，从数据库层面保证了引脚不会重复，同时保留了扩展空间，缺点是需要额外维护这个辅助表，增加了一点复杂度。

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