将列表转换为查询集
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>)
为了提高性能,在我的项目中,大多数模型实例都作为列表值存储在缓存中。但是 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>)