python - Django 休息框架 : advanced queryset filtering based on another model

Question

我正在使用 Django Rest Framework 作为应用程序的后端。

我有一个用户，它有一个钱包。然后我有 Item。如果 User 想要一个 Item，它会在他/她的 Wallet 中创建一个名为 WalletItem 的实例。一切正常。

现在我想使用属性 limit_usage 限制 User 的项目数。

首先，我在添加新实例的 post 方法中添加了一个检查，用于检查 User 的 Wallet 中项目实例的数量。因此，如果此 Item 的 limit_usage == 2，则用户在尝试添加第三个 WalletItem 时会收到 403。

我想覆盖 list()/retrieve() 中的 get_queryset() 方法或 queryset > 方法，这样如果匿名用户调用 /items/ 就会有未过滤的项目作为响应。但是，如果用户经过身份验证，我只想过滤那些他/她被允许放入钱包的项目，即那些没有超过当前用户 limit_usage 的项目。

class Wallet(models.Model):
    user = models.OneToOneField('auth.User', related_name='wallet')

class Item(models.Model):
    valid_from = models.DateTimeField()
    valid_to = models.DateTimeField()
    limit_usage = models.PositiveSmallIntegerField(default=0)

class WalletItem(models.Model):
    wallet = models.ForeignKey('Wallet', related_name='%(class)ss')
    offer = models.ForeignKey('Item', related_name='offer')

class ItemViewSet(viewsets.ReadOnlyModelViewSet):
    queryset = Item.objects.all().order_by('-created_at')
    serializer_class = ItemSerializer

    def list(self, request, *args, **kwargs):
        time_now = now()
        self.queryset = self.queryset.filter(
            valid_from__lte=time_now,
            valid_to__gte=time_now,
        )
        serializer = self.get_serializer(self.queryset, many=True)
        return Response(serializer.data)

我创建了一个 Item 类的方法，它应该对我有帮助，但我意识到我不能在查询集中使用它:

def is_visible_for_user(self, user=None):
    if not self.limit_usage or not user:
        return True
    ct = WalletItem.objects.filter(item=self, wallet=user.wallet).count()
    return self.limit_usage > ct

所以我可以遍历查询集以查看用户是否可以看到每个项目，但是我无法从这个过滤列表中构建查询集。我在 SO 上发现了类似的东西:Django REST Framework : filtering with another table但回应对我没有帮助。

Answer 1

您首先需要检查用户是否通过身份验证，如果没有，然后返回每个 Item。如果相应的WalletItem对象超过了限制，则过滤掉Item对象。

from django.db.models import Count, F, Sum

...

class ItemViewSet(viewsets.ReadOnlyModelViewSet):
    def get_queryset(self):
        queryset = super().get_queryset()
        user = self.request.user
        if user.is_anonymous:
            return queryset

        queryset = queryset.annotate(user_wallet_items=Sum(
            Case(
                When(walletitem__wallet_id=user.wallet_id, then=1),
                default=0, output_field=IntegerField()
            )) \
            .filter(user_wallet_items__lte=F('limit_usage'))
        return queryset

我建议您将基于当前时间的过滤移动到相同的 get_queryset() 方法，因为它属于那里。

注意:我没有测试过这种方法。