Django Query优化：将21次查询缩减为2次，按站点配置获取Article列表

完全可以实现仅2次查询完成需求，同时还能解决原有实现pk__in不保证返回顺序的隐性问题。

实现思路 #

1. 第一次查询：拉取所有激活站点的配置，仅需1次查询，记录每个站点的排序优先级和单站点返回文章上限
2. 第二次查询：借助Django内置的窗口函数，给同站点下的文章按创建时间倒序排名，过滤掉排名超过上限的文章，最后按站点优先级+文章创建时间排序，仅需1次查询

优化后代码 #

from django.db.models import Window, F, Case, When, IntegerField
from django.db.models.functions import RowNumber

def home(self) -> QuerySet:
    # 第一次查询：获取所有激活站点的配置
    active_sites = WebSite.objects.active().order_by('home_order', 'name')
    site_order = {site.pk: idx for idx, site in enumerate(active_sites)}
    site_article_limit = {site.pk: site.home_article_count for site in active_sites}

    # 第二次查询：批量拉取符合要求的文章
    return self.get_queryset() \
        .filter(is_active=True, website_id__in=site_order.keys()) \
        .annotate(
            # 同站点内文章按创建时间倒序排名
            article_rank_in_site=Window(
                expression=RowNumber(),
                partition_by=F('website_id'),
                order_by=F('created').desc()
            ),
            # 注入站点排序权重，用于全局排序
            site_sort_priority=Case(
                *[When(website_id=site_id, then=sort_val) for site_id, sort_val in site_order.items()],
                output_field=IntegerField()
            ),
            # 注入站点文章上限，用于过滤
            max_article_count=Case(
                *[When(website_id=site_id, then=limit) for site_id, limit in site_article_limit.items()],
                output_field=IntegerField()
            )
        ) \
        # 过滤掉超出单站点上限的文章
        .filter(article_rank_in_site__lte=F('max_article_count')) \
        # 按站点优先级+文章创建时间倒序排序，完全符合需求顺序
        .order_by('site_sort_priority', '-created')

注：以上代码默认Article模型中指向WebSite的外键字段名为website，如果你的实际字段名不同请对应修改。

拓展：1次查询实现（仅PostgreSQL兼容） #

如果你的项目使用PostgreSQL，可以借助CTE将两次查询合并为1次，性能会更高：

from django.db.models import Window, F, Subquery, OuterRef, IntegerField
from django.db.models.functions import RowNumber

def home(self) -> QuerySet:
    active_sites = WebSite.objects.active().order_by('home_order', 'name')
    return self.get_queryset() \
        .filter(is_active=True, website_id__in=Subquery(active_sites.values('pk'))) \
        .annotate(
            article_rank_in_site=Window(
                expression=RowNumber(),
                partition_by=F('website_id'),
                order_by=F('created').desc()
            ),
            site_sort_priority=Subquery(
                active_sites.filter(pk=OuterRef('website_id')).values('home_order')[:1],
                output_field=IntegerField()
            ),
            max_article_count=Subquery(
                active_sites.filter(pk=OuterRef('website_id')).values('home_article_count')[:1],
                output_field=IntegerField()
            )
        ) \
        .filter(article_rank_in_site__lte=F('max_article_count')) \
        .order_by('site_sort_priority', '-created')

注意事项 #

• 窗口函数支持Django 2.0及以上版本，主流数据库MySQL 8.0+、PostgreSQL 10+、SQLite 3.25+均原生支持
• 原有实现返回的QuerySet不会按你期望的站点顺序返回，因为pk__in查询不会遵循传入的主键列表顺序，优化后的实现直接解决了这个问题

内容的提问来源于stack exchange，提问作者Milano