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

我正在使用 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但回应对我没有帮助。

请您参考如下方法:

您首先需要检查用户是否通过身份验证，如果没有，然后返回每个 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() 方法，因为它属于那里。

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