如何在Flask应用中实现类似first_or_404()的first_or_redirect()方法

我开发了一个集成Flask-Login的Flask应用，希望防止用户访问其他用户的数据。目前用下面的代码会直接返回404错误页面，但我想改成重定向到首页：

@login_required
def myurl(po_id):
    current_po = Purchaseorder.query.filter_by(id=po_id, owner=current_user.id)\
        .first_or_404()
    # 删除、编辑等操作逻辑

我也试过另一种写法：

@login_required
def myurl(po_id):
    current_po = Purchaseorder.query.filter_by(id=po_id).first()
    if current_po.owner!=current_user.id:
        redirect ...

但每次都重复写这类判断太繁琐，有没有简便的实现方案？另外能不能支持.all()的查询场景？

解决方案 #

1. 自定义装饰器处理单条数据权限 #

写一个通用装饰器，统一处理单条数据的用户权限校验，避免重复编写判断逻辑：

from functools import wraps
from flask import redirect, url_for
from flask_login import current_user

def user_owns_model(model_class, id_arg_name='po_id', owner_field='owner'):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # 从路由参数中提取目标数据ID
            model_id = kwargs.get(id_arg_name)
            if not model_id:
                return redirect(url_for('index'))
            
            # 查询目标数据实例
            model_instance = model_class.query.get(model_id)
            # 校验：数据不存在 或 当前用户不是数据所有者，直接重定向首页
            if not model_instance or getattr(model_instance, owner_field) != current_user.id:
                return redirect(url_for('index'))
            
            # 把校验通过的实例传给视图函数，省去重复查询
            kwargs['model_instance'] = model_instance
            return f(*args, **kwargs)
        return decorated_function
    return decorator

在视图中直接复用这个装饰器：

@login_required
@user_owns_model(Purchaseorder, id_arg_name='po_id')
def myurl(po_id, model_instance):
    # 直接使用model_instance即可，无需再次查询
    # 执行删除、编辑等操作逻辑

这个装饰器支持适配任意模型，只要传入对应模型类、路由参数名、所有者字段名即可。

2. 处理.all()的批量数据场景 #

针对批量查询当前用户数据的场景，可以通过模型类方法或通用查询函数简化代码：

方式一：给模型添加类方法

class Purchaseorder(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    owner = db.Column(db.Integer, db.ForeignKey('user.id'))
    # ...其他字段定义

    @classmethod
    def get_user_orders(cls, user_id):
        return cls.query.filter_by(owner=user_id).all()

视图中直接调用：

@login_required
def my_orders():
    user_orders = Purchaseorder.get_user_orders(current_user.id)
    # 处理订单列表逻辑

方式二：通用查询函数（适配多模型）

def get_user_models(model_class, user_id, owner_field='owner'):
    return model_class.query.filter_by(**{owner_field: user_id}).all()

使用时：

@login_required
def my_orders():
    user_orders = get_user_models(Purchaseorder, current_user.id)
    # 处理逻辑

额外优化：复杂批量查询的复用 #

如果需要带过滤、排序的复杂批量查询，可以先构建当前用户的基础查询对象，再叠加条件：

@login_required
def filtered_orders(status):
    # 先获取当前用户的基础查询对象
    base_query = Purchaseorder.query.filter_by(owner=current_user.id)
    # 叠加过滤、排序条件
    filtered_orders = base_query.filter_by(status=status)\
        .order_by(Purchaseorder.create_time.desc()).all()
    # 处理逻辑

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