将列表转换为查询集

Convert list into queryset

为了提高性能,在我的项目中,大多数模型实例都作为列表值存储在缓存中。但是 Django Rest Framework 中的所有通用视图都希望它们是查询集对象。如何将从列表中获得的值转换为对象等查询集,以便我可以使用通用视图。

说,我有一个像

这样的函数
def cache_user_articles(user_id):
    key = "articles_{0}".format(user_id)

    articles = cache.get(key)

    if articles is None:
        articles = list(Article.objects.filter(user_id = user_id))
        cache.set(key, articles)

    return articles

在我的 views.py,

class ArticleViewSet(viewsets.ModelViewSet):

    ...

    def get_queryset(self, request, *args, **kwargs):
        return cache_user_articles(kwargs.get(user_id))

但是,这当然不会起作用,因为 Django Rest Framework 期望 get_queryset 的结果是 QuerySet 对象,并且在 PUT 请求时它会调用 'get' 方法在上面。有什么办法,我可以让它与通用 DRF 视图一起工作。

由于 Duck Typing,这就是 Python 类动态语言真正闪耀的地方。你可以很容易地写出像 QuerySet.

这样嘎嘎作响的东西
import mongoengine
from bson import ObjectId


class DuckTypedQuerySet(list):

    def __init__(self, data, document):
        if not hasattr(data, '__iter__') or isinstance(data, mongoengine.Document):
            raise TypeError("DuckTypedQuerySet requires iterable data")

        super(DuckTypedQuerySet, self).__init__(data)

        self._document = document

    @property
    def objects(self):
        return self

    def _query_match(self, instance, **kwargs):
        is_match = True

        for key, value in kwargs.items():
            attribute = getattr(instance, key, None)

            if isinstance(attribute, ObjectId) and not isinstance(value, ObjectId):
                attribute = str(attribute)

            if not attribute == value:
                is_match = False
                break

        return is_match


    def filter(self, **kwargs):
        data = filter(lambda instance: self._query_match(instance, **kwargs), self)

        return self.__class__(data, self._document)

    def get(self, **kwargs):
        results = self.filter(**kwargs)

        if len(results) > 1:
            raise self._document.MultipleObjectsReturned("{0} items returned, instead of 1".format(len(results)))

        if len(results) < 1:
            raise self._document.DoesNotExist("{0} matching query does not exist.".format(str(self._document)))

        return results[0]

    def first(self):
        return next(iter(self), None)

    def all(self):
        return self

    def count(self):
        return len(self)


def cache_user_articles(user_id):
    key = "articles_{0}".format(user_id)

    articles = cache.get(key)

    if articles is None:
        articles = DuckTypedQuerySet(list(Article.objects.filter(user_id = user_id)), document = Article)
        cache.set(key, articles)

    return articles

当然,这并不是一个详尽的实现。您可能需要添加查询集中存在的其他方法。但我认为这些将适用于简单的用例。现在您可以使用 Django Rest Framework 的通用实现。

这个呢? (我用 class 变量模拟了 redis 缓存)

class CachedManager(models.Manager):

    cache = dict()

    def cached(self, user_id):
        cached = self.cache.get(user_id, [])
        if not cached:
            self.cache[user_id] = [article.pk for article in self.filter(user_id=user_id)]

        return self.cache[user_id]


class Article(models.Model):

    objects = CachedManager()

    user_id = models.IntegerField()
    #  Whatever fields your Article model has

然后在您的视图中或您需要的任何地方:

你可以打电话给Article.objects.cached(<a user id>)