filter/exclude 中的一对多最新查询

Question

如果我有：

class Info(Model):
    ...

class Ad(Model):
    listed_date = DatetimeField()
    info = ForeignKey('Info', related_name='ads', null=True)
    ....

我想根据 Ad 中的字段查询 Info，但只查询最新的广告。我知道我能做到：

Ad.objects.latest('listed_date')

但是由于我将通过将几个 filter/excludes 链接在一起来构建查询，所以我想要这样的东西：

query = query.filter(
    Q(**{
        'ads__latest__'+attr_name: query_value
    })
)

或者甚至有一个字段 'latest_ad' 始终指向基于特定字段的最新信息。目标是能够在构建的 filter/exclude 方法中查询相关领域的最新信息。

我该怎么做？

编辑:

一点背景... 我有 2 个模型（LegalAd，TrusteeInfo）存储关于同一拍卖项目的抓取数据，一些领域需要公平处理以提取必要的值（因此我决定将信息存储在单独的模型中）将数据存储在加工的不同阶段。然后，我尝试将两个模型合并为一个模型 (AuctionItem)，并广泛使用属性来优先处理来自 TrusteeInfo 而非 LegalAd 的数据，以获取它们共享的类似字段。问题是我想查询那些属性禁止使用的字段。所以我创建了一个管理器并覆盖了过滤器和排除方法来保存优先级逻辑。下面是代码：

class LegalAd(models.Model):
    listed_date = models.DateField(null=True)  # field I would like to use for latest query
    auction = models.ForeignKey('auction_table.Auction', related_name='legal_ads', null=True)
    ...


class TrusteeInfo(models.Model):
    auction = models.OneToOneField('auction_table.Auction', null=True)
    ...


class AuctionManager(models.Manager):
    def do_query_action(self, action, kwargs):
        trusteeinfo = apps.get_model('scrapers', 'TrusteeInfo')
        trustee_fields = [field.name for field in trusteeinfo._meta.get_fields()]
        legalad = apps.get_model('scrapers', 'LegalAd')
        legalad_fields = [field.name for field in legalad._meta.get_fields()]
        related_fields = trustee_fields + legalad_fields
        auction_native_fields = [
            'legal_ads',
            'trusteeinfo',
            'properties',
            'id',
            'pk',
            'created_date',
            'updated_date'
        ]
        query = super(AuctionManager, self)
        for attr, value in kwargs.items():
            attr_base = attr.split('__')[0]  # get the base attr name
            if attr_base in auction_native_fields:
                query = getattr(query, action)(**{attr: value})
            elif attr_base in related_fields:
                qs = []
                if attr_base in trustee_fields:
                    trustee_attr_name = 'trusteeinfo__' + attr
                    qs.append(Q(**{trustee_attr_name: value}))
                if attr_base in legalad_fields:
                    legalad_attr_name = 'legalads__' + attr
                    qs.append(Q(**{legalad_attr_name: value}))
                query = getattr(query, action)(reduce(or_, qs))
            else:
                raise AttributeError("type object `Auction` has no attribute '{attr}'".format(attr=attr))
        return query.distinct()

    def filter(self, **kwargs):
        return self.do_query_action('filter', kwargs)

    def exclude(self, **kwargs):
        return self.do_query_action('exclude', kwargs)


class Auction(models.Model):
    objects = AuctionManager()
    created_date = models.DateTimeField(auto_now_add=True)
    updated_date = models.DateTimeField(auto_now=True)

    @property
    def latest_ad(self):
        return self.legal_ads.exists() and self.legal_ads.latest('listed_date')

    @property
    def sale_datetime(self):
        if self.trusteeinfo and self.trusteeinfo.sale_datetime:
            return self.trusteeinfo.sale_datetime
        else:
            return self.latest_ad and self.latest_ad.sale_datetime

    @property
    def county(self):
        if self.trusteeinfo and self.trusteeinfo.county:
            return self.trusteeinfo.county
        else:
            return self.latest_ad and self.latest_ad.county

    @property
    def sale_location(self):
        return self.latest_ad and self.latest_ad.sale_address

    @property
    def property_addresses(self):
        if self.trusteeinfo and self.trusteeinfo.parsed_addresses.exists():
            return self.trusteeinfo.parsed_addresses
        else:
            return self.latest_ad and self.latest_ad.parsed_addresses

    @property
    def raw_addresses(self):
        if self.trusteeinfo and self.trusteeinfo.addresses:
            return self.trusteeinfo.addresses
        else:
            return self.latest_ad and self.latest_ad.addresses.get('addresses', None)

    @property
    def parcel_numbers(self):
        return self.latest_ad and self.latest_ad.parcel_numbers

    @property
    def trustee(self):
        if self.trusteeinfo:
            return self.trusteeinfo.trustee
        else:
            return self.latest_ad and self.latest_ad.trustee.get('trustee', None)

    @property
    def opening_bid(self):
        if self.trusteeinfo and self.trusteeinfo.opening_bid:
            return self.trusteeinfo.opening_bid
        else:
            return self.latest_ad and self.latest_ad.dollar_amounts.get('bid_owed', [[None]])[0][0]

    @property
    def deposit_amount(self):
        if self.trusteeinfo and self.trusteeinfo.deposit_amount:
            return self.trusteeinfo.deposit_amount
        else:
            return self.latest_ad and self.latest_ad.dollar_amounts.get('deposit', [[None]])[0][0]

    @property
    def sale_status(self):
        return self.trusteeinfo and self.trusteeinfo.sale_status

    @property
    def trustors(self):
        if self.trusteeinfo and self.trusteeinfo.parsed_names.exists():
            return self.trusteeinfo.parsed_names
        else:
            return self.latest_ad and self.latest_ad.parsed_names

由于广告通常一次列出 2 个，因此很可能在最新日期显示 2 个广告，这意味着我必须运行也有类似 first() 方法的东西。我可以寻找某些 kwargs 和运行一个特殊的查询，但我如何将它合并到链式查询中的其余 kwargs 中？理想情况下，如果我可以保留一对多 legal_ads，但也能够执行以下操作：

query.filter(latest_ad__<queryfield>=value)

或：

query.filter(legal_ads__latest__<queryfield>=value)

那就太好了。

Answer 1

您可以使用 .latest() 和 .filter()

Ad.objects.filter(your_filter=your_value).latest('listed_date')

或使用oder_by

Ad.objects.filter(your_filter=your_value).order_by('-listed_date')[0]

Answer 2

你遇到的是所谓的greatest-n-per-group问题，用ORM很难处理甚至不可能。

可以找到一种解决问题的方法 here。

在你的情况下可能是这样的：

Info.objects.filter(
    ad__listed_date__in=Info.objects.annotate(
            last_date=Max('ad__listed_date')
        ).values_list('last_date', flat=True)
    #now you can add more
    #ad__<somefiled> statements
    #but you need to make it in a single `.filter` call
    #otherwise the ORM will do separate joins per `.filter` call
)

我个人不喜欢这个。对我来说这看起来像是一个黑客，它不是很有效，并且如果某些组中的倒数第二个 ad 与最后一个 ad 相等 listed_date ，它很容易 return 坏结果]在另一组。

解决方法

如果您向我们提供更多背景信息，说明为什么需要根据每个信息过滤 latest_ad，也许我们可以找到另一种方法来获得 same/similar 结果。

但是，我更喜欢的一种解决方法是过滤某些 date_range。例如，不要搜索 latest_ad，而是在 last_day 或 two 或 week 中的 latest_ads 上搜索 .filter，具体取决于您的需求。它非常简单高效 （易于优化） 查询。

Info.objects.filter(
    ad__listed_date__gte=(today-timedelta(days=1))
    #now again you can keep adding more `ad__<somefiled>` statements
    #but make sure to enclose them in a single `.filter` call.
)

您还提到了一个很好的解决方法，如果您可以轻松地保持最新的 Info.latest_ad 字段，那么我想您会很高兴。

如果您采用这种方法，请务必设置 on_delete=models.SET_NULL，因为默认行为（级联删除）会给您带来问题。

class Info(Model):
    #...
    
    #related_name='+' prevents creating a reverse relation
    #on_delete=models.SET_NULL prevents deleting the Info object when its latest ad got deleted.
    latest_ad = models.ForeignKey('Ad', 
        related_name='+',
        blank=True,
        null=True,
        on_delete=models.SET_NULL
    )

filter/exclude 中的一对多最新查询

one to many latest query in a filter/exclude

django

django-models

greatest-n-per-group

解决方法