Flask-SQLAlchemy实现多条件过滤排序、用户收藏数据去重

现有数据表结构 #

你当前基于Flask-SQLAlchemy设计的三张表结构如下：

from . import db
from flask_login import UserMixin
from sqlalchemy.sql import func


class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(150), unique=True)
    password = db.Column(db.String(150))
    first_name = db.Column(db.String(150))
    anime = db.relationship('Anime')
    movie = db.relationship('Movie')


class Movie(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(50), nullable=False)
    date = db.Column(db.DateTime(timezone=True), default=func.now())
    year = db.Column(db.Integer, nullable=False)
    description = db.Column(db.String(400), nullable=False)
    rating = db.Column(db.Float, nullable=True)
    ranking = db.Column(db.Integer, nullable=True)
    review = db.Column(db.String(50), nullable=True)
    img_url = db.Column(db.String(250), nullable=False)
    movie_link = db.Column(db.String(250), nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))


class Anime(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(50), nullable=False)
    year = db.Column(db.Integer, nullable=False)
    date = db.Column(db.DateTime(timezone=True), default=func.now())
    description = db.Column(db.String(400), nullable=False)
    rating = db.Column(db.Float, nullable=True)
    ranking = db.Column(db.Integer, nullable=True)
    review = db.Column(db.String(50), nullable=True)
    img_url = db.Column(db.String(250), nullable=False)
    anime_link = db.Column(db.String(250), nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))

问题1：用户收藏动漫的过滤、排序与排名更新 #

错误原因 #

你之前的代码存在两个核心问题：

• 过滤字段写错：Anime表的id是动漫条目自身的主键，关联用户的外键是user_id，用id=current_user.id过滤完全匹配不到正确数据
• 没正确使用链式查询：Flask-SQLAlchemy支持在同一个查询语句上拼接过滤、排序、分页等逻辑，不需要拆分多次查询。

正确实现代码 #

# 查询当前登录用户的所有动漫收藏，按评分从高到低排序
all_anime = Anime.query.filter_by(user_id=current_user.id).order_by(Anime.rating.desc()).all()

# 遍历更新排名，评分最高的条目排名为1
for idx, anime in enumerate(all_anime):
    anime.ranking = idx + 1
    # 如果需要倒序排名（评分越低排名越靠前），把上一行替换为 anime.ranking = len(all_anime) - idx

# 提交所有修改到数据库
db.session.commit()

如果需要按评分升序排列，把.desc()去掉即可，排名逻辑对应调整。

问题2：新增动漫时的重复收藏校验 #

错误原因 #

之前的校验逻辑只匹配了用户ID、或者只匹配了动漫标题，没有做双字段联合校验，会出现「A用户收藏过的动漫B用户无法收藏」或者「同个用户可以重复添加同名动漫」的bug。

正确实现代码 #

filter_by()中传入多个参数时，默认是AND匹配逻辑，即必须同时满足所有条件才会返回结果，刚好适配双字段校验的需求，修正后的路由代码如下：

@views.route("/anime/find")
def find_anime():
    anime_api_id = request.args.get("id")
    if anime_api_id:
        # 修正原代码中url多写了一个斜杠的问题
        anime_api_url = f"https://kitsu.io/api/edge/anime/{anime_api_id}"
        response = requests.get(anime_api_url)
        data = response.json()['data']['attributes']
        name = data["canonicalTitle"].replace(' ', '%20')
        description = data['description'].rsplit('.', 1)[0]

        # 同时匹配当前用户ID、动漫标题，检查是否已收藏
        existing_anime = Anime.query.filter_by(
            user_id=current_user.id,
            title=data['canonicalTitle']
        ).first()
        
        if existing_anime:
            flash("Anime Already exists In The List.", category="error")
            # 重复时直接跳转，终止后续新增逻辑
            return redirect(url_for("views.home"))
        
        # 校验通过才执行新增
        new_anime = Anime(
            title=data['canonicalTitle'],
            year=data['startDate'].split("-")[0],
            img_url=data['posterImage']['medium'],
            description=description,
            anime_link=ANIME_SEARCH + name,
            user_id=current_user.id
        )
        db.session.add(new_anime)
        db.session.commit()
        flash("Successfully Added.", category="success")
        return redirect(url_for("views.edit_anime", id=new_anime.id))

补充说明 #

• 所有查询方法（filter_by/filter/order_by/limit等）都可以链式拼接，最终调用all()/first()/count()等方法才会真正执行SQL查询
• 写过滤条件时一定要区分表自身主键和关联外键，避免字段写错导致查询结果不符合预期
• 校验到重复数据后一定要加return终止后续逻辑，否则还是会执行新增操作。

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