在 Django QuerySet 中，如何在多对一关系中过滤“不存在”答案

【问题标题】：In a Django QuerySet, how to filter for "not exists" in a many-to-one relationship在 Django QuerySet 中，如何在多对一关系中过滤“不存在”
【发布时间】：2013-01-27 16:29:36
【问题描述】：

我有两个这样的模型：

class User(models.Model):
    email = models.EmailField()

class Report(models.Model):
    user = models.ForeignKey(User)

实际上，每个模型都有更多与此问题无关的字段。

我想过滤所有拥有以“a”开头的电子邮件但没有报告的用户。将有更多基于其他字段的.filter() 和.exclude() 条件。

我想这样处理它：

users = User.objects.filter(email__like = 'a%')

users = users.filter(<other filters>)

users = ???

我想要？？？过滤掉没有与之关联的报告的用户。我该怎么做？如果这不可能如我所介绍的那样，还有什么替代方法？

【问题讨论】：

Generic solution for Django 1.11+

标签： django django-models django-orm

【解决方案1】：

注意：这个答案是在 2013 年为 Django 1.5 编写的。请参阅其他答案以了解适用于较新版本 Django 的更好方法

使用isnull。

users_without_reports = User.objects.filter(report__isnull=True)
users_with_reports = User.objects.filter(report__isnull=False).distinct()

当您使用isnull=False 时，需要distinct() 以防止重复结果。

【讨论】：

这没问题，但在 __isnull=True 和 __isnull=False 的情况下，它会生成带有 report 的 OUTER JOIN。对于有报告的用户的问题，它的效率可能低于INNER JOIN。对于这种情况，我发现了一个丑陋的黑客：User.objects.filter(report__id__gt=0).distinct()。这假设 ID > 0，这不一定是个案。有没有更好的强制内部连接的方法，有人吗？
@OrangeDog 解释一下为什么它更好会很有用，即使这就像“使用 NOT EXISTS （通常？）比进行连接更有效”一样简单
@OrangeDog 公平点，但我希望你能确认它是否能提供更好的性能。鉴于该问题不包含任何 SQL，我不确定 OP 是否特别想要使用 NOT EXISTS 进行查询。
@Alasdair 是的，使用WHERE NOT EXISTS 几乎总是比使用DISTINCT ... OUTER JOIN 更快，因为它避免了可能巨大的中间结果集。

【解决方案2】：

从 Django 3.0 开始，您现在可以直接在 filter() 中使用表达式，删除不必要的 SQL 子句：

User.objects.filter(
    ~Exists(Reports.objects.filter(user__eq=OuterRef('pk'))),
    email__startswith='a'
)

SELECT user.pk, user.email
FROM user
WHERE NOT EXISTS (SELECT U0.pk FROM reports U0 WHERE U0.user = user.pk) AND email LIKE 'a%';

文档：

对于 Django 1.11+，您可以添加 EXISTS 子查询：

User.objects.annotate(
    no_reports=~Exists(Reports.objects.filter(user__eq=OuterRef('pk')))
).filter(
    email__startswith='a',
    no_reports=True
)

这会生成类似这样的 SQL：

SELECT
    user.pk,
    user.email,
    NOT EXISTS (SELECT U0.pk FROM reports U0 WHERE U0.user = user.pk) AS no_reports
FROM user
WHERE email LIKE 'a%' AND NOT EXISTS (SELECT U0.pk FROM reports U0 WHERE U0.user = user.pk);

NOT EXISTS 子句几乎总是执行“不存在”过滤器的最有效方法。

【讨论】：

这应该是最佳答案。谢谢@OrangeDog。

【解决方案3】：

在没有额外查询或 JOIN 的情况下获得原生 SQL EXISTS/NOT EXISTS 的唯一方法是将其作为原始 SQL 添加到 .extra() 子句中：

users = users.extra(where=[
    """NOT EXISTS(SELECT 1 FROM {reports} 
                  WHERE user_id={users}.id)
    """.format(reports=Report._meta.db_table, users=User._meta.db_table)
])

事实上，这是一个非常明显和有效的解决方案，我有时想知道为什么它没有内置到 Django 作为查找。它还允许细化子查询以查找例如只有在上周有[out]报告的用户，或者有[out]未答复/未查看的报告的用户。

【讨论】：

@OrangeDog 参数不能用于传递表名。它们将被数据库引擎引用。此外，额外保护表名也没有用，因为它们来自代码，而不是来自用户输入。
引用表名有什么问题？根据它们是什么，它们可能需要被引用。这是您应该使用参数的原因之一。
@OrangeDog - 因为大多数数据库引擎将单引号解释为字符串文字 - 表需要双引号/角度引用，或者根本不引用

【解决方案4】：

Alasdair's answer 很有帮助，但我不喜欢使用distinct()。它有时会很有用，但通常是代码异味告诉您您搞砸了联接。

幸运的是，Django 的queryset 允许您过滤子查询。使用 Django 3.0，您还可以使用 exists clause。

以下是从您的问题中运行查询的几种方法：

# Tested with Django 3.0 and Python 3.6
import logging
import sys

import django
from django.apps import apps
from django.apps.config import AppConfig
from django.conf import settings
from django.db import connections, models, DEFAULT_DB_ALIAS
from django.db.models import Exists, OuterRef
from django.db.models.base import ModelBase

NAME = 'udjango'
DB_FILE = NAME + '.db'


def main():
    setup()

    class User(models.Model):
        email = models.EmailField()

        def __repr__(self):
            return 'User({!r})'.format(self.email)

    class Report(models.Model):
        user = models.ForeignKey(User, on_delete=models.CASCADE)

    syncdb(User)
    syncdb(Report)

    anne = User.objects.create(email='anne@example.com')
    User.objects.create(email='adam@example.com')
    alice = User.objects.create(email='alice@example.com')
    User.objects.create(email='bob@example.com')

    Report.objects.create(user=anne)
    Report.objects.create(user=alice)
    Report.objects.create(user=alice)

    logging.info('users without reports')
    logging.info(User.objects.filter(report__isnull=True, email__startswith='a'))

    logging.info('users with reports (allows duplicates)')
    logging.info(User.objects.filter(report__isnull=False, email__startswith='a'))

    logging.info('users with reports (no duplicates)')
    logging.info(User.objects.exclude(report__isnull=True).filter(email__startswith='a'))

    logging.info('users with reports (no duplicates, simpler SQL)')
    report_user_ids = Report.objects.values('user_id')
    logging.info(User.objects.filter(id__in=report_user_ids, email__startswith='a'))

    logging.info('users with reports (EXISTS clause, Django 3.0)')
    logging.info(User.objects.filter(
        Exists(Report.objects.filter(user_id=OuterRef('id'))),
        email__startswith='a'))

    logging.info('Done.')


def setup():
    with open(DB_FILE, 'w'):
        pass  # wipe the database
    settings.configure(
        DEBUG=True,
        DATABASES={
            DEFAULT_DB_ALIAS: {
                'ENGINE': 'django.db.backends.sqlite3',
                'NAME': DB_FILE}},
        LOGGING={'version': 1,
                 'disable_existing_loggers': False,
                 'formatters': {
                    'debug': {
                        'format': '%(asctime)s[%(levelname)s]'
                                  '%(name)s.%(funcName)s(): %(message)s',
                        'datefmt': '%Y-%m-%d %H:%M:%S'}},
                 'handlers': {
                    'console': {
                        'level': 'DEBUG',
                        'class': 'logging.StreamHandler',
                        'formatter': 'debug'}},
                 'root': {
                    'handlers': ['console'],
                    'level': 'INFO'},
                 'loggers': {
                    "django.db": {"level": "DEBUG"}}})
    app_config = AppConfig(NAME, sys.modules['__main__'])
    apps.populate([app_config])
    django.setup()
    original_new_func = ModelBase.__new__

    @staticmethod
    def patched_new(cls, name, bases, attrs):
        if 'Meta' not in attrs:
            class Meta:
                app_label = NAME
            attrs['Meta'] = Meta
        return original_new_func(cls, name, bases, attrs)
    ModelBase.__new__ = patched_new


def syncdb(model):
    """ Standard syncdb expects models to be in reliable locations.

    Based on https://github.com/django/django/blob/1.9.3
    /django/core/management/commands/migrate.py#L285
    """
    connection = connections[DEFAULT_DB_ALIAS]
    with connection.schema_editor() as editor:
        editor.create_model(model)


main()

如果你把它放到一个 Python 文件中并运行它，你应该会看到如下内容：

2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) PRAGMA foreign_keys = OFF; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) PRAGMA foreign_keys; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) BEGIN; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.schema.execute(): CREATE TABLE "udjango_user" ("id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, "email" varchar(254) NOT NULL); (params None)
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) CREATE TABLE "udjango_user" ("id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, "email" varchar(254) NOT NULL); args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) PRAGMA foreign_key_check; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) PRAGMA foreign_keys = ON; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) PRAGMA foreign_keys = OFF; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) PRAGMA foreign_keys; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) BEGIN; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.schema.execute(): CREATE TABLE "udjango_report" ("id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, "user_id" integer NOT NULL REFERENCES "udjango_user" ("id") DEFERRABLE INITIALLY DEFERRED); (params None)
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) CREATE TABLE "udjango_report" ("id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, "user_id" integer NOT NULL REFERENCES "udjango_user" ("id") DEFERRABLE INITIALLY DEFERRED); args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) PRAGMA foreign_key_check; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.schema.execute(): CREATE INDEX "udjango_report_user_id_60bc619c" ON "udjango_report" ("user_id"); (params ())
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) CREATE INDEX "udjango_report_user_id_60bc619c" ON "udjango_report" ("user_id"); args=()
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) PRAGMA foreign_keys = ON; args=None
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.017) INSERT INTO "udjango_user" ("email") VALUES ('anne@example.com'); args=['anne@example.com']
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.023) INSERT INTO "udjango_user" ("email") VALUES ('adam@example.com'); args=['adam@example.com']
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.022) INSERT INTO "udjango_user" ("email") VALUES ('alice@example.com'); args=['alice@example.com']
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.022) INSERT INTO "udjango_user" ("email") VALUES ('bob@example.com'); args=['bob@example.com']
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.029) INSERT INTO "udjango_report" ("user_id") VALUES (1); args=[1]
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.033) INSERT INTO "udjango_report" ("user_id") VALUES (3); args=[3]
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.033) INSERT INTO "udjango_report" ("user_id") VALUES (3); args=[3]
2019-12-06 11:45:17[INFO]root.main(): users without reports
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) SELECT "udjango_user"."id", "udjango_user"."email" FROM "udjango_user" LEFT OUTER JOIN "udjango_report" ON ("udjango_user"."id" = "udjango_report"."user_id") WHERE ("udjango_user"."email" LIKE 'a%' ESCAPE '\' AND "udjango_report"."id" IS NULL) LIMIT 21; args=('a%',)
2019-12-06 11:45:17[INFO]root.main(): <QuerySet [User('adam@example.com')]>
2019-12-06 11:45:17[INFO]root.main(): users with reports (allows duplicates)
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) SELECT "udjango_user"."id", "udjango_user"."email" FROM "udjango_user" INNER JOIN "udjango_report" ON ("udjango_user"."id" = "udjango_report"."user_id") WHERE ("udjango_user"."email" LIKE 'a%' ESCAPE '\' AND "udjango_report"."id" IS NOT NULL) LIMIT 21; args=('a%',)
2019-12-06 11:45:17[INFO]root.main(): <QuerySet [User('anne@example.com'), User('alice@example.com'), User('alice@example.com')]>
2019-12-06 11:45:17[INFO]root.main(): users with reports (no duplicates)
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) SELECT "udjango_user"."id", "udjango_user"."email" FROM "udjango_user" WHERE (NOT ("udjango_user"."id" IN (SELECT U0."id" FROM "udjango_user" U0 LEFT OUTER JOIN "udjango_report" U1 ON (U0."id" = U1."user_id") WHERE U1."id" IS NULL)) AND "udjango_user"."email" LIKE 'a%' ESCAPE '\') LIMIT 21; args=('a%',)
2019-12-06 11:45:17[INFO]root.main(): <QuerySet [User('anne@example.com'), User('alice@example.com')]>
2019-12-06 11:45:17[INFO]root.main(): users with reports (no duplicates, simpler SQL)
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) SELECT "udjango_user"."id", "udjango_user"."email" FROM "udjango_user" WHERE ("udjango_user"."email" LIKE 'a%' ESCAPE '\' AND "udjango_user"."id" IN (SELECT U0."user_id" FROM "udjango_report" U0)) LIMIT 21; args=('a%',)
2019-12-06 11:45:17[INFO]root.main(): <QuerySet [User('anne@example.com'), User('alice@example.com')]>
2019-12-06 11:45:17[INFO]root.main(): users with reports (EXISTS clause, Django 3.0)
2019-12-06 11:45:17[DEBUG]django.db.backends.debug_sql(): (0.000) SELECT "udjango_user"."id", "udjango_user"."email" FROM "udjango_user" WHERE (EXISTS(SELECT U0."id", U0."user_id" FROM "udjango_report" U0 WHERE U0."user_id" = "udjango_user"."id") AND "udjango_user"."email" LIKE 'a%' ESCAPE '\') LIMIT 21; args=('a%',)
2019-12-06 11:45:17[INFO]root.main(): <QuerySet [User('anne@example.com'), User('alice@example.com')]>
2019-12-06 11:45:17[INFO]root.main(): Done.

可以看到最终查询使用了所有的内连接。

【讨论】：

您的代码 sn-p 看起来可能不完整。最后一条语句是Report.objects.create(user=anne)，而您的输出和想法表明您打算显示更多代码。我很想看！
不确定你在说什么，@KrystianCybulski。我的浏览器显示了这么多代码，但是有一个滚动条可以查看其余部分。如果您遇到浏览器问题，请尝试编辑答案以查看 Markdown 源代码。
你是对的。我道歉。 Mac 上的 Chrome 很聪明，将滚动条隐藏在嵌入式代码框中。我没有意识到它是可滚动的。
我的朋友们，这才是真正的艺术——不是现代的“我明白（购买|明白）它而你不明白”艺术，而是原始的、您的工艺精通的巧妙展示，艺术。这不是一招一式的鱼，而是带有独立测试池的钓鱼竿，任何人都可以轻松地演示如何投掷鱼线，或尝试一些快速的捕虫诱饵。干得好，唐。

【解决方案5】：

除了@OrangeDog 的回答。 Since Django 3.0 你可以使用Exists 子查询直接过滤一个查询集：

User.objects.filter(
    ~Exists(Reports.objects.filter(user__eq=OuterRef('pk'))
)

【讨论】：

【解决方案6】：

要过滤没有与之关联的报告的用户，请尝试以下操作：

users = User.objects.exclude(id__in=[elem.user.id for elem in Report.objects.all()])

【讨论】：

这是有希望的。我需要看看这会生成什么样的 SQL。我还认为id__in = Report.objects.all() 足以满足.exclude() 的调用。
如果你只使用users=User.objects.exclude(id__in=Report.objects.all())，你会得到所有id与任何报告id相同的用户
这是一个糟糕的答案，除非您只打算自己使用该网站，或者在您的密友中使用该网站；) 如果您有 10 亿份报告，这可能会使数据库崩溃。经验法则：永远不要在查询中使用__in=！此外，您甚至没有Report.objects.values('user_id').distinct()，这对于报告多但用户少的情况会有所帮助。
我最好改进我的“经验法则”——只有当你有固定数量的可能性时才使用something__in=[my_value1, my_value2, ...]。这实际上意味着您以某种方式手动指定它。 OKAY - 如果你传递一个 queryset 对象，这实际上会变成一个 SQL 子查询，比如：WHERE "auth_user"."id" in (SELECT U0."id" FROM "report" U0)。所以你仍然可以希望数据库能理解它。
对于阅读此答案的人：[elem.user.id for elem in Report.objects.all()] 效率极低 - 它从报告中获取所有字段（当只需要一个时），无缘无故地构建所有 Report 实例，最后做一个每个报告的额外 sql 查询以检索用户...list(Report.objects.values_list('user_id', flat=True)) 将解决所有这些问题（仍然不是最好的解决方案，但是...）。

【解决方案7】：

查找存在连接行的行的最佳选择：

Report.objects.filter(user__isnull=False).distinct()

这使用INNER JOIN（然后冗余检查User.id 不为空）。

查找没有连接行的行的最佳选择：

Report.objects.filter(user__isnull=True)

这使得LEFT OUTER JOIN，然后检查User.id 不为空。

基于连接的查询将比子查询更快，因此这比 Django >= 3 等新可用选项更快，用于查找行没有连接行：

Report.objects.filter(~Exists(User.objects.filter(report=OuterRef('pk'))))

这会创建一个WHERE NOT EXISTS (SELECT .. FROM User..)，因此涉及一个可能很大的中间结果集（感谢@Tomasz Gandor）。

这对于 Django filter() 不能传递子查询，也使用子查询所以速度较慢：

Report.objects.annotate(
    no_users=~Exists(User.objects.filter(report=OuterRef('pk')))
).filter(no_users=True)

这可以与子查询结合使用。在这个例子中，Textbook 有多个Versions（即version 有textbook_id），version 有多个Pages（即page 有version_id） .子查询获取关联页面的每本教科书的最新版本：

subquery = (
    Version.objects
        .filter(
            # OuterRef joins to Version.textbook in outer query below
            textbook=OuterRef('textbook'), 
            # excludes rows with no joined Page records
            page__isnull=False)
        # ordered so [:1] below gets highest (ie, latest) version number
        .order_by('-number').distinct()
)
# Only the Version.ids of the latest versions that have pages returned by the subquery
books = Version.objects.filter(pk=Subquery(subquery.values('pk')[:1])).distinct()

要返回连接到两个表中的一个或两个表的行，请使用 Q 对象（Page 和 TextMarkup 都具有连接到 File 的可空外键）：

from django.db.models import Q

File.objects.filter(Q(page__isnull=False) | Q(textmarkup__isnull=False).distinct()

【讨论】：

查找存在 连接行的行的最佳选择：

查找没有连接行的行的最佳选择：

查找存在连接行的行的最佳选择：