在不破坏 DRY 的情况下自定义 QuerySet 和 Manager？

Posted 2023-02-23

【中文标题】在不破坏 DRY 的情况下自定义 QuerySet 和 Manager？

【英文标题】：Custom QuerySet and Manager without breaking DRY?

【发布时间】：2011-01-10 21:54:38

【问题描述】：

我正在尝试找到一种方法来实现自定义 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 中实现它，但我最终复制了我所有的方法定义。

我还发现 this snippet 有效，但我需要将额外的参数传递给 for_user 所以它会崩溃，因为它严重依赖于重新定义 get_query_set。

有没有办法在不重新定义QuerySet 和Manager 子类中的所有方法的情况下做到这一点？

【问题讨论】：

警告：当使用 .defer 或 .only 方法时，T.Stone 选择的答案会导致严重的性能损失（从毫秒响应时间到多秒响应）。例如，在 Django 1.3 中，如下查询： MyModel.objects.only('some_field').get(id=1) => 在 3.7 毫秒内返回，但是，如上所述添加 CustomManager，我得到： MyModel.objects .only('some_field').get(id=1) => 在 ~ 357 毫秒内返回

有其他人复制过这个吗？ Django 1.4 怎么样？

好的。但是为什么以及如何发生这种情况？查询是否不同，或者您是否在没有实际访问数据库的情况下分析了该操作？

【参考方案1】：

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

更多详情请见Creating Manager with QuerySet methods。

【讨论】：

这是最好的方法，但它必须举例说明for_user 方法应该如何获取用户并返回self.[...] 以将多个操作链接在一起。

有趣的是，这不是他第一次在这里回答。从专业角度来说，这是最干净的解决方案。

【参考方案2】：

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)

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

class CustomQuerySetManager(models.Manager):

到

from django.contrib.auth.models import UserManager

class CustomQuerySetManager(UserManager):
    ***

【讨论】：

你能决定如何处理***.com/edit-suggestions/1216

Stone，您将不得不自己编辑它，被拒绝后无法进行编辑

警告：我尝试了这种方法，发现它严重减慢了 .defer 和 .only 的调用速度。

我已将此回复更新到社区 wiki。有性能优化的可以根据需要调整代码。

可能类似于来自pypi.python.org/pypi/django-model-utils的PassThroughManager

【参考方案3】：

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

【参考方案4】：

您现在可以在管理器上使用from_queryset() 方法来更改其基本查询集。

这允许您只定义一次查询集方法和管理器方法

来自文档

对于高级用法，您可能需要自定义管理器和自定义查询集。您可以通过调用 Manager.from_queryset() 来做到这一点，它会返回一个带有自定义 QuerySet 方法副本的基础 Manager 的子类：

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

【讨论】：

我喜欢你的解决方案！因为自定义查询集中定义的所有方法都可供经理使用。 :-)

【参考方案5】：

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)

【讨论】：

太棒了，他们在 django 本身就有这样的装饰器

我的 Django 希望 get_queryset 被覆盖，而不是 get_query_set。

【参考方案6】：

在某些用例中，我们需要 call custom QuerySet methods from the manager 而不是使用 QuerySet 的 get_manager 方法。

根据一个已接受的解决方案 cmets 中发布的解决方案，一个 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)

通过上面的方法，我们可以像下面这样访问相关对象（书），而无需在管理器中为每个查询集方法定义新方法。

fiction_books = author.books.fiction()

【参考方案7】：

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

【参考方案8】：

以下对我有用。

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

但没有理由不适合二级经理。

【讨论】：

尝试使用filter，然后使用get_active_for_account。它在您的示例中有效，但在您已经使用filter，然后正在使用QuerySet（这是我的示例）之后就不行了。