Django ORM:是否可以注入子查询?
Django ORM: is it possible to inject subqueries?
我有一个看起来像这样的 Django 模型:
class Result(models.Model):
date = DateTimeField()
subject = models.ForeignKey('myapp.Subject')
test_type = models.ForeignKey('myapp.TestType')
summary = models.PositiveSmallIntegerField()
# more fields about the result like its location, tester ID and so on
有时我们想要检索所有测试结果,有时我们只想要每个主题的特定测试类型的最新结果。 This answer 为 SQL 提供了一些很棒的选项,可以找到最新的结果。
此外,有时我们希望将结果分为不同的时间块,以便我们可以绘制出每天/每周/每月的结果数量。
我们还想过滤各种字段,为了优雅起见,我想要一个 QuerySet,然后我可以进行所有 filter() 调用,并对计数进行注释,而不是制作原始 SQL 来电。
我已经走到这一步了:
qs = Result.objects.extra(select = {
'date_range': "date_trunc('{0}', time)".format("day"), # Chunking into time buckets
'rn' : "ROW_NUMBER() OVER(PARTITION BY subject_id, test_type_id ORDER BY time DESC)"})
qs = qs.values('date_range', 'result_summary', 'rn')
qs = qs.order_by('-date_range')
结果如下 SQL:
SELECT (ROW_NUMBER() OVER(PARTITION BY subject_id, test_type_id ORDER BY time DESC)) AS "rn", (date_trunc('day', time)) AS "date_range", "myapp_result"."result_summary" FROM "myapp_result" ORDER BY "date_range" DESC
这有点接近我想要的,但现在我需要以某种方式过滤以仅获取 rn = 1 的行。我尝试在 extra() 中使用 'where' 字段,这给出了我以下 SQL 和错误:
SELECT (ROW_NUMBER() OVER(PARTITION BY subject_id, test_type_id ORDER BY time DESC)) AS "rn", (date_trunc('day', time)) AS "date_range", "myapp_result"."result_summary" FROM "myapp_result" WHERE "rn"=1 ORDER BY "date_range" DESC ;
ERROR: column "rn" does not exist
所以我认为找到 "rn" 的查询需要是一个子查询——但是否有可能以某种方式做到这一点,也许使用 extra()?
我知道我可以使用原始 SQL 来做到这一点,但它看起来很难看!我很想找到一个很好的简洁方法,我有一个可过滤的查询集。
我想另一个选择是在模型中有一个字段,指示它是否实际上是该主题的该测试类型的最新结果...
我找到方法了!
qs = Result.objects.extra(where = ["NOT EXISTS(SELECT * FROM myapp_result as T2 WHERE (T2.test_type_id = myapp_result.test_type_id AND T2.subject_id = myapp_result.subject ID AND T2.time > myapp_result.time))"])
这是基于与 the answer I referenced earlier 不同的选项。我可以随心所欲地过滤或注释 qs。
顺便说一句,在解决这个问题的过程中我尝试了这个:
qq = Result.objects.extra(where = ["NOT EXISTS(SELECT * FROM myapp_result as T2 WHERE (T2.test_type_id = myapp_result.test_type_id AND T2.subject_id = myapp_result.subject ID AND T2.time > myapp_result.time))"])
qs = Result.objects.filter(id__in=qq)
Django 可以按照您的意愿嵌入子查询:
SELECT ...some fields... FROM "myapp_result"
WHERE ("myapp_result"."id" IN (SELECT "myapp_result"."id" FROM "myapp_result"
WHERE (NOT EXISTS(SELECT * FROM myapp_result as T2
WHERE (T2.subject_id = myapp_result.subject_id AND T2.test_type_id = myapp_result.test_type_id AND T2.time > myapp_result.time)))))
我意识到这有比我需要的更多的子查询,但我在这里注意到它,因为我可以想象知道你可以用一个查询集过滤另一个查询集并且 Django 完全按照你希望的方式做嵌入子查询(而不是执行它并嵌入返回值,这会很糟糕。)
我有一个看起来像这样的 Django 模型:
class Result(models.Model):
date = DateTimeField()
subject = models.ForeignKey('myapp.Subject')
test_type = models.ForeignKey('myapp.TestType')
summary = models.PositiveSmallIntegerField()
# more fields about the result like its location, tester ID and so on
有时我们想要检索所有测试结果,有时我们只想要每个主题的特定测试类型的最新结果。 This answer 为 SQL 提供了一些很棒的选项,可以找到最新的结果。
此外,有时我们希望将结果分为不同的时间块,以便我们可以绘制出每天/每周/每月的结果数量。
我们还想过滤各种字段,为了优雅起见,我想要一个 QuerySet,然后我可以进行所有 filter() 调用,并对计数进行注释,而不是制作原始 SQL 来电。
我已经走到这一步了:
qs = Result.objects.extra(select = {
'date_range': "date_trunc('{0}', time)".format("day"), # Chunking into time buckets
'rn' : "ROW_NUMBER() OVER(PARTITION BY subject_id, test_type_id ORDER BY time DESC)"})
qs = qs.values('date_range', 'result_summary', 'rn')
qs = qs.order_by('-date_range')
结果如下 SQL:
SELECT (ROW_NUMBER() OVER(PARTITION BY subject_id, test_type_id ORDER BY time DESC)) AS "rn", (date_trunc('day', time)) AS "date_range", "myapp_result"."result_summary" FROM "myapp_result" ORDER BY "date_range" DESC
这有点接近我想要的,但现在我需要以某种方式过滤以仅获取 rn = 1 的行。我尝试在 extra() 中使用 'where' 字段,这给出了我以下 SQL 和错误:
SELECT (ROW_NUMBER() OVER(PARTITION BY subject_id, test_type_id ORDER BY time DESC)) AS "rn", (date_trunc('day', time)) AS "date_range", "myapp_result"."result_summary" FROM "myapp_result" WHERE "rn"=1 ORDER BY "date_range" DESC ;
ERROR: column "rn" does not exist
所以我认为找到 "rn" 的查询需要是一个子查询——但是否有可能以某种方式做到这一点,也许使用 extra()?
我知道我可以使用原始 SQL 来做到这一点,但它看起来很难看!我很想找到一个很好的简洁方法,我有一个可过滤的查询集。
我想另一个选择是在模型中有一个字段,指示它是否实际上是该主题的该测试类型的最新结果...
我找到方法了!
qs = Result.objects.extra(where = ["NOT EXISTS(SELECT * FROM myapp_result as T2 WHERE (T2.test_type_id = myapp_result.test_type_id AND T2.subject_id = myapp_result.subject ID AND T2.time > myapp_result.time))"])
这是基于与 the answer I referenced earlier 不同的选项。我可以随心所欲地过滤或注释 qs。
顺便说一句,在解决这个问题的过程中我尝试了这个:
qq = Result.objects.extra(where = ["NOT EXISTS(SELECT * FROM myapp_result as T2 WHERE (T2.test_type_id = myapp_result.test_type_id AND T2.subject_id = myapp_result.subject ID AND T2.time > myapp_result.time))"])
qs = Result.objects.filter(id__in=qq)
Django 可以按照您的意愿嵌入子查询:
SELECT ...some fields... FROM "myapp_result"
WHERE ("myapp_result"."id" IN (SELECT "myapp_result"."id" FROM "myapp_result"
WHERE (NOT EXISTS(SELECT * FROM myapp_result as T2
WHERE (T2.subject_id = myapp_result.subject_id AND T2.test_type_id = myapp_result.test_type_id AND T2.time > myapp_result.time)))))
我意识到这有比我需要的更多的子查询,但我在这里注意到它,因为我可以想象知道你可以用一个查询集过滤另一个查询集并且 Django 完全按照你希望的方式做嵌入子查询(而不是执行它并嵌入返回值,这会很糟糕。)