python - Flask-sqlalchemy 检查约束 - 一列的值不能与另一列的任何值相同？

Question

class Sensor(db.Model):

    __table_args__ = (
        db.CheckConstraint('dat_pin != clk_pin', name='Data Pin != 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, unique=True, nullable=False,)
    clk_pin  = db.Column(db.Integer, unique=True, 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) #not a column in table, just backreference


    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}')"

我需要设置一个约束，以便 dat_pin 或 clock_pin 永远不能与表中的任何其他 dat_pin 或 clk_pin 相同。我试图添加一个表范围的约束，但这似乎只是检查我试图添加的行。我还需要它来检查任何以前的记录，例如：

DAT_PIN CLK_PIN
1       2       = OK (Because both are different from each other and different from any other dat_pin or clock_pin records)
DAT_PIN CLK_PIN
3       3       = NOT OK (This is how it works currently)
DAT_PIN CLK_PIN
4       1       = NOT OK (Because the first record has dat_pin=1)
DAT_PIN CLK_PIN
2       5       = ALSO NOT OK (Because clk_pin has already been 2)

我希望这个例子足够清楚，我很难用一句话来定义这个约束，英语不是我的第一语言。此外，我需要其他列不受约束，例如 full_val 仍然可以具有与任何 dat_pin 或 clk_pin 相同的值。

标签： pythonsqlitesqlalchemyflask-sqlalchemy

Answer 1

如果您将传感器和引脚之间的关系存储在另一个表中，并为每个记录指定引脚类型的鉴别器字段，那么您可以使用引脚的编号作为主键，这将强制两者之间的数字的唯一性引脚类型。我们还可以使用内置的 sqlalchemy 继承模式来自动设置鉴别器字段，并让我们为每种类型的 pin 使用漂亮的描述性类名称。

也许最好让代码来说话：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

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


class Sensor(db.Model):

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, nullable=False,)
    dat_pin = db.relationship(
        "Pin",
        primaryjoin="and_(Sensor.id==Pin.sensor_id, Pin.type=='dat')",
        uselist=False,
    )
    clk_pin = db.relationship(
        "Pin",
        primaryjoin="and_(Sensor.id==Pin.sensor_id, Pin.type=='clk')",
        uselist=False,
    )


class Pin(db.Model):

    id = db.Column(db.Integer, primary_key=True)
    type = db.Column(db.Enum("dat", "clk"), nullable=False)
    sensor_id = db.Column(db.Integer, db.ForeignKey("sensor.id"))

    __table_args__ = (db.UniqueConstraint("type", "sensor_id"),)
    __mapper_args__ = {"polymorphic_on": type}


class DatPin(Pin):

    __mapper_args__ = {"polymorphic_identity": "dat"}


class ClkPin(Pin):

    __mapper_args__ = {"polymorphic_identity": "clk"}


if __name__ == "__main__":
    # proof of concept
    with app.test_request_context():
        db.drop_all()
        db.create_all()
        sensor = Sensor(name="sensor")
        db.session.add(sensor)
        db.session.commit()
        sensor.dat_pin = DatPin(id=1)
        sensor.clk_pin = ClkPin(id=2)
        db.session.commit()
        print(sensor.dat_pin, sensor.clk_pin)

    # add dat_pin with id=1 to a sensor,
    # and then a clk_pin with id=1 to another sensor
    with app.test_request_context():
        db.drop_all()
        db.create_all()
        sensor1 = Sensor(name="sensor1")
        sensor2 = Sensor(name="sensor2")
        db.session.add_all([sensor1, sensor2])
        db.session.commit()
        sensor1.dat_pin = DatPin(id=1)
        sensor2.clk_pin = ClkPin(id=1)
        # will raise integrity error due to multiple primary keys
        try:
            db.session.commit()
        except Exception as e:
            print("Yep, raised %s", e)
            db.session.rollback()

    # test to prove only one type of each pin allowed per sensor
    with app.test_request_context():
        db.drop_all()
        db.create_all()
        db.session.add(Sensor(name="sensor", id=1))
        db.session.add_all([ClkPin(sensor_id=1), ClkPin(sensor_id=1)])
        # will raise integrity error due to violation of unique constraint
        try:
            db.session.commit()
        except Exception as e:
            print("Yep, raised %s", e)