我正在尝试找到一种方法来实现自定义
QuerySet
和自定义 Manager
而不破坏 DRY。这是我到目前为止所拥有的:
class MyInquiryManager(models.Manager):
def for_user(self, user):
return self.get_query_set().filter(
Q(assigned_to_user=user) |
Q(assigned_to_group__in=user.groups.all())
)
class Inquiry(models.Model):
ts = models.DateTimeField(auto_now_add=True)
status = models.ForeignKey(InquiryStatus)
assigned_to_user = models.ForeignKey(User, blank=True, null=True)
assigned_to_group = models.ForeignKey(Group, blank=True, null=True)
objects = MyInquiryManager()
这工作得很好,直到我做了这样的事情:
inquiries = Inquiry.objects.filter(status=some_status)
my_inquiry_count = inquiries.for_user(request.user).count()
这会立即破坏所有内容,因为
QuerySet
没有与 Manager
相同的方法。我尝试创建一个自定义 QuerySet
类,并在 MyInquiryManager
中实现它,但最终我复制了所有方法定义。
我还发现这个片段可以工作,但是我需要将额外的参数传递给
for_user
,所以它会崩溃,因为它严重依赖于重新定义get_query_set
。
有没有一种方法可以做到这一点,而无需在
QuerySet
和 Manager
子类中重新定义所有方法?
Django 1.7 发布了一种新且简单的方式来创建组合查询集和模型管理器:
class InquiryQuerySet(models.QuerySet):
def for_user(self, user):
return self.filter(
Q(assigned_to_user=user) |
Q(assigned_to_group__in=user.groups.all())
)
class Inquiry(models.Model):
objects = InqueryQuerySet.as_manager()
请参阅使用 QuerySet 方法创建管理器了解更多详细信息。
Django 已更改! 在使用这个写于 2009 年的答案中的代码之前,请务必查看其余答案和 Django 文档,看看是否有更合适的解决方案。
我实现这一点的方法是添加实际的
get_active_for_account
作为自定义 QuerySet
的方法。然后,为了使其在管理器下工作,您可以简单地捕获 __getattr__
并相应地返回它
为了使该模式可重复使用,我将
Manager
位提取到单独的模型管理器中:
custom_queryset/models.py
from django.db import models
from django.db.models.query import QuerySet
class CustomQuerySetManager(models.Manager):
"""A re-usable Manager to access a custom QuerySet"""
def __getattr__(self, attr, *args):
try:
return getattr(self.__class__, attr, *args)
except AttributeError:
# don't delegate internal methods to the queryset
if attr.startswith('__') and attr.endswith('__'):
raise
return getattr(self.get_query_set(), attr, *args)
def get_query_set(self):
return self.model.QuerySet(self.model, using=self._db)
一旦你得到了这个,在你的模型上你需要做的就是定义一个
QuerySet
作为自定义内部类并将管理器设置为你的自定义管理器:
your_app/models.py
from custom_queryset.models import CustomQuerySetManager
from django.db.models.query import QuerySet
class Inquiry(models.Model):
objects = CustomQuerySetManager()
class QuerySet(QuerySet):
def active_for_account(self, account, *args, **kwargs):
return self.filter(account=account, deleted=False, *args, **kwargs)
使用此模式,其中任何一个都可以工作:
>>> Inquiry.objects.active_for_account(user)
>>> Inquiry.objects.all().active_for_account(user)
>>> Inquiry.objects.filter(first_name='John').active_for_account(user)
UPD如果您与自定义用户一起使用它(
AbstractUser
),您需要更改class CustomQuerySetManager(models.Manager):
到
from django.contrib.auth.models import UserManager
class CustomQuerySetManager(UserManager):
***
您可以使用 mixin 在管理器和查询集上提供方法。
这也避免了使用
__getattr__()
方法。
from django.db.models.query import QuerySet
class PostMixin(object):
def by_author(self, user):
return self.filter(user=user)
def published(self):
return self.filter(published__lte=datetime.now())
class PostQuerySet(QuerySet, PostMixin):
pass
class PostManager(models.Manager, PostMixin):
def get_query_set(self):
return PostQuerySet(self.model, using=self._db)
您现在可以在管理器上使用 from_queryset() 方法来更改其基本查询集。
这允许您仅定义一次查询集方法和管理器方法
来自文档
对于高级用法,您可能需要自定义管理器和自定义查询集。您可以通过调用 Manager.from_queryset() 来做到这一点,它返回基 Manager 的子类以及自定义 QuerySet 方法的副本:
class InqueryQueryset(models.Queryset):
def custom_method(self):
""" available on all default querysets"""
class BaseMyInquiryManager(models.Manager):
def for_user(self, user):
return self.get_query_set().filter(
Q(assigned_to_user=user) |
Q(assigned_to_group__in=user.groups.all())
)
MyInquiryManager = BaseInquiryManager.from_queryset(InquiryQueryset)
class Inquiry(models.Model):
ts = models.DateTimeField(auto_now_add=True)
status = models.ForeignKey(InquiryStatus)
assigned_to_user = models.ForeignKey(User, blank=True, null=True)
assigned_to_group = models.ForeignKey(Group, blank=True, null=True)
objects = MyInquiryManager()
基于
django 3.1.3
源代码,我找到了一个简单的解决方案
from django.db.models.manager import BaseManager
class MyQuerySet(models.query.QuerySet):
def my_custom_query(self):
return self.filter(...)
class MyManager(BaseManager.from_queryset(MyQuerySet)):
...
class MyModel(models.Model):
objects = MyManager()
T. Stone 方法的稍微改进版本:
def objects_extra(mixin_class):
class MixinManager(models.Manager, mixin_class):
class MixinQuerySet(QuerySet, mixin_class):
pass
def get_query_set(self):
return self.MixinQuerySet(self.model, using=self._db)
return MixinManager()
类装饰器使使用变得简单:
class SomeModel(models.Model):
...
@objects_extra
class objects:
def filter_by_something_complex(self, whatever parameters):
return self.extra(...)
...
更新:支持非标准 Manager 和 QuerySet 基类,例如。 G。 @objects_extra(django.contrib.gis.db.models.GeoManager,django.contrib.gis.db.models.query.GeoQuerySet):
def objects_extra(Manager=django.db.models.Manager, QuerySet=django.db.models.query.QuerySet):
def oe_inner(Mixin, Manager=django.db.models.Manager, QuerySet=django.db.models.query.QuerySet):
class MixinManager(Manager, Mixin):
class MixinQuerySet(QuerySet, Mixin):
pass
def get_query_set(self):
return self.MixinQuerySet(self.model, using=self._db)
return MixinManager()
if issubclass(Manager, django.db.models.Manager):
return lambda Mixin: oe_inner(Mixin, Manager, QuerySet)
else:
return oe_inner(Mixin=Manager)
在某些用例中,我们需要 从管理器调用自定义 QuerySet 方法,而不是使用 QuerySet 的
get_manager
方法。
根据已接受的解决方案评论之一中发布的解决方案,混合就足够了。
class CustomQuerySetManagerMixin:
"""
Allow Manager which uses custom queryset to access queryset methods directly.
"""
def __getattr__(self, name):
# don't delegate internal methods to queryset
# NOTE: without this, Manager._copy_to_model will end up calling
# __getstate__ on the *queryset* which causes the qs (as `all()`)
# to evaluate itself as if it was being pickled (`len(self)`)
if name.startswith('__'):
raise AttributeError
return getattr(self.get_queryset(), name)
例如,
class BookQuerySet(models.QuerySet):
def published(self):
return self.filter(published=True)
def fiction(self):
return self.filter(genre="fiction")
def non_fiction(self):
return self.filter(genre="non-fiction")
class BookManager(CustomQuerySetManagerMixin, models.Manager):
def get_queryset(self):
return BookQuerySet(self.model, using=self._db).published()
class Book(models.Model):
title = models.CharField(max_length=200)
genre = models.CharField(choices=[('fiction', _('Fiction')), ('non-fiction', _('Non-Fiction'))])
published = models.BooleanField(default=False)
author = models.ForeignKey(Author, on_delete=models.CASCADE, related_name="books")
objects = BookManager()
class Author(models.Model):
name = models.CharField(max_length=200)
通过上述内容,我们可以像下面一样访问相关对象(Book),而无需在管理器中为每个查询集方法定义新方法。
fiction_books = author.books.fiction()
以下内容对我有用。
def get_active_for_account(self,account,*args,**kwargs):
"""Returns a queryset that is
Not deleted
For the specified account
"""
return self.filter(account = account,deleted=False,*args,**kwargs)
这是在默认管理器上;所以我曾经做过类似的事情:
Model.objects.get_active_for_account(account).filter()
但是没有理由它不适合二级经理。