Flask自引用多对多关系中如何避免重复引用

问题背景 #

开发Flask笔记管理应用，笔记间支持相互引用，已实现自引用多对多关系和禁止自引用逻辑，但无法正确检测重复引用——当尝试添加已存在的引用时，无法触发提示并跳转。

现有代码 #

模型代码 #

from datetime import datetime
from app import db

references = db.Table('references',
    db.Column('refers_id', db.Integer, db.ForeignKey('note.id')),
    db.Column('is_referred_id', db.Integer, db.ForeignKey('note.id')))

class Note(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(140), index=True, unique=True)
    timestampCr = db.Column(db.DateTime, index=True, default=datetime.utcnow)
    content = db.Column(db.Text)
    external_source = db.Column(db.String(300))
    internal_document = db.Column(db.String(300))
    timestampRe = db.Column(db.DateTime, index=True, default=datetime.utcnow)
    timestampUp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
    category = db.Column(db.Boolean, default=False, nullable=False)
    referred = db.relationship(
        'Note', secondary=references,
        primaryjoin=(references.c.refers_id == id),
        secondaryjoin=(references.c.is_referred_id == id),
        backref=db.backref('references', lazy='dynamic'), lazy='dynamic')
  
    def __repr__(self):
        return '[Note] {}' .format(self.title)

视图代码 #

@app.route("/addref/<id>", methods=['GET', 'POST'])
def addref(id):
    form = RefNoteForm()
    noteref = Note.query.filter_by(id=id).first_or_404()
    if form.validate_on_submit():
       refnote = form.opts.data
       if refnote==noteref: 
           flash ('A note can not refer to itself')
           return redirect(url_for('readnote', id=id))
       noteref.referred.append(refnote)
       db.session.commit()
       return redirect(url_for('readnote', id=id))
    return render_template('addref.html', form=form, id=id)

错误尝试分析 #

之前尝试用if noteref.referred(refnote):检测重复，这是错误语法——因为referred设置了lazy='dynamic'，返回的是SQLAlchemy查询对象，而非可直接调用的方法或列表，无法通过这种方式判断引用是否存在。

正确解决方案 #

方案1：数据库层面查询检测（高效） #

利用动态关系的查询接口，直接查询目标引用是否已存在：

@app.route("/addref/<id>", methods=['GET', 'POST'])
def addref(id):
    form = RefNoteForm()
    noteref = Note.query.filter_by(id=id).first_or_404()
    if form.validate_on_submit():
       refnote = form.opts.data
       # 禁止自引用
       if refnote == noteref: 
           flash('A note can not refer to itself')
           return redirect(url_for('readnote', id=id))
       # 检测重复引用
       if noteref.referred.filter_by(id=refnote.id).first():
           flash('This reference already exists')
           return redirect(url_for('readnote', id=id))
       # 添加新引用
       noteref.referred.append(refnote)
       db.session.commit()
       flash('Reference added successfully')
       return redirect(url_for('readnote', id=id))
    return render_template('addref.html', form=form, id=id)

方案2：转换为列表后检测（适合数据量小的场景） #

如果笔记数量不多，可以将动态查询结果转为列表后用in判断：

# 替换重复检测部分
if refnote in noteref.referred.all():
    flash('This reference already exists')
    return redirect(url_for('readnote', id=id))

额外优化：数据库层面约束（彻底防止重复） #

为了从根源避免重复引用，可以在关联表references上添加唯一约束，确保同一对refers_id和is_referred_id不会重复：

references = db.Table('references',
    db.Column('refers_id', db.Integer, db.ForeignKey('note.id')),
    db.Column('is_referred_id', db.Integer, db.ForeignKey('note.id')),
    # 添加联合唯一约束
    db.UniqueConstraint('refers_id', 'is_referred_id', name='unique_reference_pair')
)

添加约束后，若尝试插入重复引用，SQLAlchemy会抛出IntegrityError，可以在视图中捕获该异常并提示用户：

from sqlalchemy.exc import IntegrityError

# 在视图的commit部分添加异常捕获
try:
    noteref.referred.append(refnote)
    db.session.commit()
    flash('Reference added successfully')
except IntegrityError:
    db.session.rollback()
    flash('This reference already exists')

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