不破坏 DRY 的自定义查询集和管理器? ？

我试图找到一种方法来实现一个自定义的 QuerySet和一个自定义的 Manager而不破坏干燥。以下是我目前掌握的情况:

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子类中的所有方法的情况下做到这一点？

47668 次浏览
小开

下面这些对我有用。

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()

但对于二级经理来说，这没有理由不适用。

小开
最佳答案

姜戈变了！在使用2009年编写的这个答案中的代码之前，一定要查看其余的答案和 Django 文档，看看是否有更合适的解决方案。

我实现这一点的方法是添加实际的 get_active_for_account作为自定义 QuerySet的方法。然后，为了让它在管理器上工作，您可以简单地捕获 __getattr__并相应地返回它

为了使这个模式可重用，我将 Manager位提取到一个单独的模型管理器:

Custom_ queryset/model. 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/mods.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)

如果您与自定义用户(AbstractUser)一起使用 UPD，则需要进行更改
来自

class CustomQuerySetManager(models.Manager):

到

from django.contrib.auth.models import UserManager


class CustomQuerySetManager(UserManager):
***
小开

T · 斯通方法的一个略有改进的版本:

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 基类的支持，例如@Objects _ Ultra (django.Contrib.gis.db.model. GeoManager，django.Contrib.gis.db.mods.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)
小开

可以使用 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)
小开

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 方法创建 Manager。

小开

现在可以在管理器上使用 From _ queryset ()方法更改其基础查询集。

这使您只能定义一次 Queryset 方法和 Manager 方法

从医生那里

对于高级用法，您可能需要同时使用自定义 Manager 和自定义 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()
小开

在一些用例中，我们需要使用 从管理器调用自定义 QuerySet 方法而不是使用 QuerySet 的 get_manager方法。

基于已接受的解决方案评论中提供的解决方案，使用 Mixin 就足够了。

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()
小开

基于 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()

提交答案