mirror of https://github.com/django/django.git
1127 lines
39 KiB
Python
1127 lines
39 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
from unittest import mock
|
|
|
|
from django.db import connection, transaction
|
|
from django.db.models import (
|
|
BooleanField,
|
|
Case,
|
|
Count,
|
|
DecimalField,
|
|
ExpressionWrapper,
|
|
F,
|
|
FilteredRelation,
|
|
Q,
|
|
Sum,
|
|
Value,
|
|
When,
|
|
)
|
|
from django.db.models.functions import Concat
|
|
from django.db.models.lookups import Exact, IStartsWith
|
|
from django.test import TestCase
|
|
from django.test.testcases import skipUnlessDBFeature
|
|
|
|
from .models import (
|
|
Author,
|
|
Book,
|
|
BookDailySales,
|
|
Borrower,
|
|
Currency,
|
|
Editor,
|
|
ExchangeRate,
|
|
RentalSession,
|
|
Reservation,
|
|
Seller,
|
|
)
|
|
|
|
|
|
class FilteredRelationTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.author1 = Author.objects.create(name="Alice")
|
|
cls.author2 = Author.objects.create(name="Jane")
|
|
cls.editor_a = Editor.objects.create(name="a")
|
|
cls.editor_b = Editor.objects.create(name="b")
|
|
cls.book1 = Book.objects.create(
|
|
title="Poem by Alice",
|
|
editor=cls.editor_a,
|
|
author=cls.author1,
|
|
)
|
|
cls.book1.generic_author.set([cls.author2])
|
|
cls.book2 = Book.objects.create(
|
|
title="The book by Jane A",
|
|
editor=cls.editor_b,
|
|
author=cls.author2,
|
|
)
|
|
cls.book3 = Book.objects.create(
|
|
title="The book by Jane B",
|
|
editor=cls.editor_b,
|
|
author=cls.author2,
|
|
)
|
|
cls.book4 = Book.objects.create(
|
|
title="The book by Alice",
|
|
editor=cls.editor_a,
|
|
author=cls.author1,
|
|
)
|
|
cls.author1.favorite_books.add(cls.book2)
|
|
cls.author1.favorite_books.add(cls.book3)
|
|
|
|
def test_select_related(self):
|
|
qs = (
|
|
Author.objects.annotate(
|
|
book_join=FilteredRelation("book"),
|
|
)
|
|
.select_related("book_join__editor")
|
|
.order_by("pk", "book_join__pk")
|
|
)
|
|
with self.assertNumQueries(1):
|
|
self.assertQuerySetEqual(
|
|
qs,
|
|
[
|
|
(self.author1, self.book1, self.editor_a, self.author1),
|
|
(self.author1, self.book4, self.editor_a, self.author1),
|
|
(self.author2, self.book2, self.editor_b, self.author2),
|
|
(self.author2, self.book3, self.editor_b, self.author2),
|
|
],
|
|
lambda x: (x, x.book_join, x.book_join.editor, x.book_join.author),
|
|
)
|
|
|
|
def test_select_related_multiple(self):
|
|
qs = (
|
|
Book.objects.annotate(
|
|
author_join=FilteredRelation("author"),
|
|
editor_join=FilteredRelation("editor"),
|
|
)
|
|
.select_related("author_join", "editor_join")
|
|
.order_by("pk")
|
|
)
|
|
self.assertQuerySetEqual(
|
|
qs,
|
|
[
|
|
(self.book1, self.author1, self.editor_a),
|
|
(self.book2, self.author2, self.editor_b),
|
|
(self.book3, self.author2, self.editor_b),
|
|
(self.book4, self.author1, self.editor_a),
|
|
],
|
|
lambda x: (x, x.author_join, x.editor_join),
|
|
)
|
|
|
|
def test_select_related_with_empty_relation(self):
|
|
qs = (
|
|
Author.objects.annotate(
|
|
book_join=FilteredRelation("book", condition=Q(pk=-1)),
|
|
)
|
|
.select_related("book_join")
|
|
.order_by("pk")
|
|
)
|
|
self.assertSequenceEqual(qs, [self.author1, self.author2])
|
|
|
|
def test_select_related_foreign_key(self):
|
|
qs = (
|
|
Book.objects.annotate(
|
|
author_join=FilteredRelation("author"),
|
|
)
|
|
.select_related("author_join")
|
|
.order_by("pk")
|
|
)
|
|
with self.assertNumQueries(1):
|
|
self.assertQuerySetEqual(
|
|
qs,
|
|
[
|
|
(self.book1, self.author1),
|
|
(self.book2, self.author2),
|
|
(self.book3, self.author2),
|
|
(self.book4, self.author1),
|
|
],
|
|
lambda x: (x, x.author_join),
|
|
)
|
|
|
|
@skipUnlessDBFeature("has_select_for_update", "has_select_for_update_of")
|
|
def test_select_related_foreign_key_for_update_of(self):
|
|
with transaction.atomic():
|
|
qs = (
|
|
Book.objects.annotate(
|
|
author_join=FilteredRelation("author"),
|
|
)
|
|
.select_related("author_join")
|
|
.select_for_update(of=("self",))
|
|
.order_by("pk")
|
|
)
|
|
with self.assertNumQueries(1):
|
|
self.assertQuerySetEqual(
|
|
qs,
|
|
[
|
|
(self.book1, self.author1),
|
|
(self.book2, self.author2),
|
|
(self.book3, self.author2),
|
|
(self.book4, self.author1),
|
|
],
|
|
lambda x: (x, x.author_join),
|
|
)
|
|
|
|
def test_without_join(self):
|
|
self.assertCountEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
),
|
|
[self.author1, self.author2],
|
|
)
|
|
|
|
def test_with_join(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
).filter(book_alice__isnull=False),
|
|
[self.author1],
|
|
)
|
|
|
|
def test_with_exclude(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
).exclude(book_alice__isnull=False),
|
|
[self.author2],
|
|
)
|
|
|
|
def test_with_join_and_complex_condition(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book",
|
|
condition=Q(
|
|
Q(book__title__iexact="poem by alice")
|
|
| Q(book__state=Book.RENTED)
|
|
),
|
|
),
|
|
).filter(book_alice__isnull=False),
|
|
[self.author1],
|
|
)
|
|
|
|
def test_internal_queryset_alias_mapping(self):
|
|
queryset = Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
).filter(book_alice__isnull=False)
|
|
self.assertIn(
|
|
"INNER JOIN {} book_alice ON".format(
|
|
connection.ops.quote_name("filtered_relation_book")
|
|
),
|
|
str(queryset.query),
|
|
)
|
|
|
|
def test_multiple(self):
|
|
qs = (
|
|
Author.objects.annotate(
|
|
book_title_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__contains="Alice")
|
|
),
|
|
book_title_jane=FilteredRelation(
|
|
"book", condition=Q(book__title__icontains="Jane")
|
|
),
|
|
)
|
|
.filter(name="Jane")
|
|
.values("book_title_alice__title", "book_title_jane__title")
|
|
)
|
|
empty = "" if connection.features.interprets_empty_strings_as_nulls else None
|
|
self.assertCountEqual(
|
|
qs,
|
|
[
|
|
{
|
|
"book_title_alice__title": empty,
|
|
"book_title_jane__title": "The book by Jane A",
|
|
},
|
|
{
|
|
"book_title_alice__title": empty,
|
|
"book_title_jane__title": "The book by Jane B",
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_with_multiple_filter(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_editor_a=FilteredRelation(
|
|
"book",
|
|
condition=Q(
|
|
book__title__icontains="book", book__editor_id=self.editor_a.pk
|
|
),
|
|
),
|
|
).filter(book_editor_a__isnull=False),
|
|
[self.author1],
|
|
)
|
|
|
|
def test_multiple_times(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_title_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__icontains="alice")
|
|
),
|
|
)
|
|
.filter(book_title_alice__isnull=False)
|
|
.filter(book_title_alice__isnull=False)
|
|
.distinct(),
|
|
[self.author1],
|
|
)
|
|
|
|
def test_exclude_relation_with_join(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=~Q(book__title__icontains="alice")
|
|
),
|
|
)
|
|
.filter(book_alice__isnull=False)
|
|
.distinct(),
|
|
[self.author2],
|
|
)
|
|
|
|
def test_with_m2m(self):
|
|
qs = Author.objects.annotate(
|
|
favorite_books_written_by_jane=FilteredRelation(
|
|
"favorite_books",
|
|
condition=Q(favorite_books__in=[self.book2]),
|
|
),
|
|
).filter(favorite_books_written_by_jane__isnull=False)
|
|
self.assertSequenceEqual(qs, [self.author1])
|
|
|
|
def test_with_m2m_deep(self):
|
|
qs = Author.objects.annotate(
|
|
favorite_books_written_by_jane=FilteredRelation(
|
|
"favorite_books",
|
|
condition=Q(favorite_books__author=self.author2),
|
|
),
|
|
).filter(favorite_books_written_by_jane__title="The book by Jane B")
|
|
self.assertSequenceEqual(qs, [self.author1])
|
|
|
|
def test_with_m2m_multijoin(self):
|
|
qs = (
|
|
Author.objects.annotate(
|
|
favorite_books_written_by_jane=FilteredRelation(
|
|
"favorite_books",
|
|
condition=Q(favorite_books__author=self.author2),
|
|
)
|
|
)
|
|
.filter(favorite_books_written_by_jane__editor__name="b")
|
|
.distinct()
|
|
)
|
|
self.assertSequenceEqual(qs, [self.author1])
|
|
|
|
def test_values_list(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
)
|
|
.filter(book_alice__isnull=False)
|
|
.values_list("book_alice__title", flat=True),
|
|
["Poem by Alice"],
|
|
)
|
|
|
|
def test_values(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
)
|
|
.filter(book_alice__isnull=False)
|
|
.values(),
|
|
[
|
|
{
|
|
"id": self.author1.pk,
|
|
"name": "Alice",
|
|
"content_type_id": None,
|
|
"object_id": None,
|
|
}
|
|
],
|
|
)
|
|
|
|
def test_extra(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
)
|
|
.filter(book_alice__isnull=False)
|
|
.extra(where=["1 = 1"]),
|
|
[self.author1],
|
|
)
|
|
|
|
@skipUnlessDBFeature("supports_select_union")
|
|
def test_union(self):
|
|
qs1 = Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
).filter(book_alice__isnull=False)
|
|
qs2 = Author.objects.annotate(
|
|
book_jane=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="the book by jane a")
|
|
),
|
|
).filter(book_jane__isnull=False)
|
|
self.assertSequenceEqual(qs1.union(qs2), [self.author1, self.author2])
|
|
|
|
@skipUnlessDBFeature("supports_select_intersection")
|
|
def test_intersection(self):
|
|
qs1 = Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
).filter(book_alice__isnull=False)
|
|
qs2 = Author.objects.annotate(
|
|
book_jane=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="the book by jane a")
|
|
),
|
|
).filter(book_jane__isnull=False)
|
|
self.assertSequenceEqual(qs1.intersection(qs2), [])
|
|
|
|
@skipUnlessDBFeature("supports_select_difference")
|
|
def test_difference(self):
|
|
qs1 = Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
).filter(book_alice__isnull=False)
|
|
qs2 = Author.objects.annotate(
|
|
book_jane=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="the book by jane a")
|
|
),
|
|
).filter(book_jane__isnull=False)
|
|
self.assertSequenceEqual(qs1.difference(qs2), [self.author1])
|
|
|
|
def test_select_for_update(self):
|
|
self.assertSequenceEqual(
|
|
Author.objects.annotate(
|
|
book_jane=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="the book by jane a")
|
|
),
|
|
)
|
|
.filter(book_jane__isnull=False)
|
|
.select_for_update(),
|
|
[self.author2],
|
|
)
|
|
|
|
def test_defer(self):
|
|
# One query for the list and one query for the deferred title.
|
|
with self.assertNumQueries(2):
|
|
self.assertQuerySetEqual(
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
)
|
|
.filter(book_alice__isnull=False)
|
|
.select_related("book_alice")
|
|
.defer("book_alice__title"),
|
|
["Poem by Alice"],
|
|
lambda author: author.book_alice.title,
|
|
)
|
|
|
|
def test_only_not_supported(self):
|
|
msg = "only() is not supported with FilteredRelation."
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
).filter(book_alice__isnull=False).select_related("book_alice").only(
|
|
"book_alice__state"
|
|
)
|
|
|
|
def test_as_subquery(self):
|
|
inner_qs = Author.objects.annotate(
|
|
book_alice=FilteredRelation(
|
|
"book", condition=Q(book__title__iexact="poem by alice")
|
|
),
|
|
).filter(book_alice__isnull=False)
|
|
qs = Author.objects.filter(id__in=inner_qs)
|
|
self.assertSequenceEqual(qs, [self.author1])
|
|
|
|
def test_nested_foreign_key(self):
|
|
qs = (
|
|
Author.objects.annotate(
|
|
book_editor_worked_with=FilteredRelation(
|
|
"book__editor",
|
|
condition=Q(book__title__icontains="book by"),
|
|
),
|
|
)
|
|
.filter(
|
|
book_editor_worked_with__isnull=False,
|
|
)
|
|
.select_related(
|
|
"book_editor_worked_with",
|
|
)
|
|
.order_by("pk", "book_editor_worked_with__pk")
|
|
)
|
|
with self.assertNumQueries(1):
|
|
self.assertQuerySetEqual(
|
|
qs,
|
|
[
|
|
(self.author1, self.editor_a),
|
|
(self.author2, self.editor_b),
|
|
(self.author2, self.editor_b),
|
|
],
|
|
lambda x: (x, x.book_editor_worked_with),
|
|
)
|
|
|
|
def test_nested_foreign_key_nested_field(self):
|
|
qs = (
|
|
Author.objects.annotate(
|
|
book_editor_worked_with=FilteredRelation(
|
|
"book__editor", condition=Q(book__title__icontains="book by")
|
|
),
|
|
)
|
|
.filter(
|
|
book_editor_worked_with__isnull=False,
|
|
)
|
|
.values(
|
|
"name",
|
|
"book_editor_worked_with__name",
|
|
)
|
|
.order_by("name", "book_editor_worked_with__name")
|
|
.distinct()
|
|
)
|
|
self.assertSequenceEqual(
|
|
qs,
|
|
[
|
|
{
|
|
"name": self.author1.name,
|
|
"book_editor_worked_with__name": self.editor_a.name,
|
|
},
|
|
{
|
|
"name": self.author2.name,
|
|
"book_editor_worked_with__name": self.editor_b.name,
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_nested_foreign_key_filtered_base_object(self):
|
|
qs = (
|
|
Author.objects.annotate(
|
|
alice_editors=FilteredRelation(
|
|
"book__editor",
|
|
condition=Q(name="Alice"),
|
|
),
|
|
)
|
|
.values(
|
|
"name",
|
|
"alice_editors__pk",
|
|
)
|
|
.order_by("name", "alice_editors__name")
|
|
.distinct()
|
|
)
|
|
self.assertSequenceEqual(
|
|
qs,
|
|
[
|
|
{"name": self.author1.name, "alice_editors__pk": self.editor_a.pk},
|
|
{"name": self.author2.name, "alice_editors__pk": None},
|
|
],
|
|
)
|
|
|
|
def test_nested_m2m_filtered(self):
|
|
qs = (
|
|
Book.objects.annotate(
|
|
favorite_book=FilteredRelation(
|
|
"author__favorite_books",
|
|
condition=Q(author__favorite_books__title__icontains="book by"),
|
|
),
|
|
)
|
|
.values(
|
|
"title",
|
|
"favorite_book__pk",
|
|
)
|
|
.order_by("title", "favorite_book__title")
|
|
)
|
|
self.assertSequenceEqual(
|
|
qs,
|
|
[
|
|
{"title": self.book1.title, "favorite_book__pk": self.book2.pk},
|
|
{"title": self.book1.title, "favorite_book__pk": self.book3.pk},
|
|
{"title": self.book4.title, "favorite_book__pk": self.book2.pk},
|
|
{"title": self.book4.title, "favorite_book__pk": self.book3.pk},
|
|
{"title": self.book2.title, "favorite_book__pk": None},
|
|
{"title": self.book3.title, "favorite_book__pk": None},
|
|
],
|
|
)
|
|
|
|
def test_nested_chained_relations(self):
|
|
qs = (
|
|
Author.objects.annotate(
|
|
my_books=FilteredRelation(
|
|
"book",
|
|
condition=Q(book__title__icontains="book by"),
|
|
),
|
|
preferred_by_authors=FilteredRelation(
|
|
"my_books__preferred_by_authors",
|
|
condition=Q(my_books__preferred_by_authors__name="Alice"),
|
|
),
|
|
)
|
|
.annotate(
|
|
author=F("name"),
|
|
book_title=F("my_books__title"),
|
|
preferred_by_author_pk=F("preferred_by_authors"),
|
|
)
|
|
.order_by("author", "book_title", "preferred_by_author_pk")
|
|
)
|
|
self.assertQuerySetEqual(
|
|
qs,
|
|
[
|
|
("Alice", "The book by Alice", None),
|
|
("Jane", "The book by Jane A", self.author1.pk),
|
|
("Jane", "The book by Jane B", self.author1.pk),
|
|
],
|
|
lambda x: (x.author, x.book_title, x.preferred_by_author_pk),
|
|
)
|
|
|
|
def test_deep_nested_foreign_key(self):
|
|
qs = (
|
|
Book.objects.annotate(
|
|
author_favorite_book_editor=FilteredRelation(
|
|
"author__favorite_books__editor",
|
|
condition=Q(author__favorite_books__title__icontains="Jane A"),
|
|
),
|
|
)
|
|
.filter(
|
|
author_favorite_book_editor__isnull=False,
|
|
)
|
|
.select_related(
|
|
"author_favorite_book_editor",
|
|
)
|
|
.order_by("pk", "author_favorite_book_editor__pk")
|
|
)
|
|
with self.assertNumQueries(1):
|
|
self.assertQuerySetEqual(
|
|
qs,
|
|
[
|
|
(self.book1, self.editor_b),
|
|
(self.book4, self.editor_b),
|
|
],
|
|
lambda x: (x, x.author_favorite_book_editor),
|
|
)
|
|
|
|
def test_relation_name_lookup(self):
|
|
msg = (
|
|
"FilteredRelation's relation_name cannot contain lookups (got "
|
|
"'book__title__icontains')."
|
|
)
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
Author.objects.annotate(
|
|
book_title=FilteredRelation(
|
|
"book__title__icontains",
|
|
condition=Q(book__title="Poem by Alice"),
|
|
),
|
|
)
|
|
|
|
def test_condition_outside_relation_name(self):
|
|
msg = (
|
|
"FilteredRelation's condition doesn't support relations outside "
|
|
"the 'book__editor' (got 'book__author__name__icontains')."
|
|
)
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
Author.objects.annotate(
|
|
book_editor=FilteredRelation(
|
|
"book__editor",
|
|
condition=Q(book__author__name__icontains="book"),
|
|
),
|
|
)
|
|
|
|
def test_condition_with_exact_lookup_outside_relation_name(self):
|
|
qs = Author.objects.annotate(
|
|
book_editor=FilteredRelation(
|
|
"book__editor",
|
|
condition=Q(book__author__name="book"),
|
|
),
|
|
).filter(book_editor__isnull=True)
|
|
self.assertEqual(qs.count(), 4)
|
|
|
|
def test_condition_with_func_and_lookup_outside_relation_name(self):
|
|
qs = Author.objects.annotate(
|
|
book_editor=FilteredRelation(
|
|
"book__editor",
|
|
condition=Q(
|
|
book__title=Concat(Value("The book by "), F("book__author__name"))
|
|
),
|
|
),
|
|
).filter(book_editor__isnull=False)
|
|
self.assertEqual(qs.count(), 1)
|
|
|
|
def test_condition_deeper_relation_name(self):
|
|
msg = (
|
|
"FilteredRelation's condition doesn't support nested relations "
|
|
"deeper than the relation_name (got "
|
|
"'book__editor__name__icontains' for 'book')."
|
|
)
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
Author.objects.annotate(
|
|
book_editor=FilteredRelation(
|
|
"book",
|
|
condition=Q(book__editor__name__icontains="b"),
|
|
),
|
|
)
|
|
|
|
def test_with_empty_relation_name_error(self):
|
|
with self.assertRaisesMessage(ValueError, "relation_name cannot be empty."):
|
|
FilteredRelation("", condition=Q(blank=""))
|
|
|
|
def test_with_condition_as_expression_error(self):
|
|
msg = "condition argument must be a Q() instance."
|
|
expression = Case(
|
|
When(book__title__iexact="poem by alice", then=True),
|
|
default=False,
|
|
)
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
FilteredRelation("book", condition=expression)
|
|
|
|
def test_with_prefetch_related(self):
|
|
msg = "prefetch_related() is not supported with FilteredRelation."
|
|
qs = Author.objects.annotate(
|
|
book_title_contains_b=FilteredRelation(
|
|
"book", condition=Q(book__title__icontains="b")
|
|
),
|
|
).filter(
|
|
book_title_contains_b__isnull=False,
|
|
)
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
qs.prefetch_related("book_title_contains_b")
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
qs.prefetch_related("book_title_contains_b__editor")
|
|
|
|
def test_with_generic_foreign_key(self):
|
|
self.assertSequenceEqual(
|
|
Book.objects.annotate(
|
|
generic_authored_book=FilteredRelation(
|
|
"generic_author", condition=Q(generic_author__isnull=False)
|
|
),
|
|
).filter(generic_authored_book__isnull=False),
|
|
[self.book1],
|
|
)
|
|
|
|
def test_eq(self):
|
|
self.assertEqual(
|
|
FilteredRelation("book", condition=Q(book__title="b")), mock.ANY
|
|
)
|
|
|
|
def test_conditional_expression(self):
|
|
qs = Author.objects.annotate(
|
|
the_book=FilteredRelation("book", condition=Q(Value(False))),
|
|
).filter(the_book__isnull=False)
|
|
self.assertSequenceEqual(qs, [])
|
|
|
|
def test_expression_outside_relation_name(self):
|
|
qs = Author.objects.annotate(
|
|
book_editor=FilteredRelation(
|
|
"book__editor",
|
|
condition=Q(
|
|
Exact(F("book__author__name"), "Alice"),
|
|
Value(True),
|
|
book__title__startswith="Poem",
|
|
),
|
|
),
|
|
).filter(book_editor__isnull=False)
|
|
self.assertSequenceEqual(qs, [self.author1])
|
|
|
|
def test_conditional_expression_with_case(self):
|
|
qs = Book.objects.annotate(
|
|
alice_author=FilteredRelation(
|
|
"author",
|
|
condition=Q(
|
|
Case(When(author__name="Alice", then=True), default=False),
|
|
),
|
|
),
|
|
).filter(alice_author__isnull=False)
|
|
self.assertCountEqual(qs, [self.book1, self.book4])
|
|
|
|
def test_conditional_expression_outside_relation_name(self):
|
|
tests = [
|
|
Q(Case(When(book__author__name="Alice", then=True), default=False)),
|
|
Q(
|
|
ExpressionWrapper(
|
|
Q(Value(True), Exact(F("book__author__name"), "Alice")),
|
|
output_field=BooleanField(),
|
|
),
|
|
),
|
|
]
|
|
for condition in tests:
|
|
with self.subTest(condition=condition):
|
|
qs = Author.objects.annotate(
|
|
book_editor=FilteredRelation("book__editor", condition=condition),
|
|
).filter(book_editor__isnull=True)
|
|
self.assertSequenceEqual(qs, [self.author2, self.author2])
|
|
|
|
def test_conditional_expression_with_lookup(self):
|
|
lookups = [
|
|
Q(book__title__istartswith="poem"),
|
|
Q(IStartsWith(F("book__title"), "poem")),
|
|
]
|
|
for condition in lookups:
|
|
with self.subTest(condition=condition):
|
|
qs = Author.objects.annotate(
|
|
poem_book=FilteredRelation("book", condition=condition)
|
|
).filter(poem_book__isnull=False)
|
|
self.assertSequenceEqual(qs, [self.author1])
|
|
|
|
def test_conditional_expression_with_expressionwrapper(self):
|
|
qs = Author.objects.annotate(
|
|
poem_book=FilteredRelation(
|
|
"book",
|
|
condition=Q(
|
|
ExpressionWrapper(
|
|
Q(Exact(F("book__title"), "Poem by Alice")),
|
|
output_field=BooleanField(),
|
|
),
|
|
),
|
|
),
|
|
).filter(poem_book__isnull=False)
|
|
self.assertSequenceEqual(qs, [self.author1])
|
|
|
|
def test_conditional_expression_with_multiple_fields(self):
|
|
qs = Author.objects.annotate(
|
|
my_books=FilteredRelation(
|
|
"book__author",
|
|
condition=Q(Exact(F("book__author__name"), F("book__author__name"))),
|
|
),
|
|
).filter(my_books__isnull=True)
|
|
self.assertSequenceEqual(qs, [])
|
|
|
|
def test_conditional_expression_rhs_contains_relation_name(self):
|
|
qs = Book.objects.annotate(
|
|
rel=FilteredRelation(
|
|
"editor",
|
|
condition=Q(id=1 * F("number_editor")),
|
|
)
|
|
).filter(rel__isnull=True)
|
|
self.assertSequenceEqual(qs, [])
|
|
|
|
def test_conditional_expression_rhs_startswith_relation_name(self):
|
|
qs = Book.objects.annotate(
|
|
rel=FilteredRelation(
|
|
"editor",
|
|
condition=Q(id=1 * F("editor_number")),
|
|
)
|
|
).filter(rel__isnull=True)
|
|
self.assertSequenceEqual(qs, [])
|
|
|
|
def test_conditional_expression_lhs_startswith_relation_name(self):
|
|
qs = Book.objects.annotate(
|
|
rel=FilteredRelation(
|
|
"editor",
|
|
condition=Q(editor_number__gt=1),
|
|
)
|
|
).filter(rel__isnull=True)
|
|
self.assertSequenceEqual(qs, [])
|
|
|
|
def test_conditional_expression_lhs_contains_relation_name(self):
|
|
qs = Book.objects.annotate(
|
|
rel=FilteredRelation(
|
|
"editor",
|
|
condition=Q(number_editor__gt=1),
|
|
)
|
|
).filter(rel__isnull=True)
|
|
self.assertSequenceEqual(qs, [])
|
|
|
|
def test_conditional_expression_does_not_support_queryset(self):
|
|
msg = "Passing a QuerySet within a FilteredRelation is not supported."
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
Author.objects.annotate(
|
|
poem_book=FilteredRelation(
|
|
"book",
|
|
condition=Q(book__in=Book.objects.filter(title__istartswith="a")),
|
|
),
|
|
).filter(poem_book__isnull=False)
|
|
|
|
|
|
class FilteredRelationAggregationTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.author1 = Author.objects.create(name="Alice")
|
|
cls.editor_a = Editor.objects.create(name="a")
|
|
cls.book1 = Book.objects.create(
|
|
title="Poem by Alice",
|
|
editor=cls.editor_a,
|
|
author=cls.author1,
|
|
)
|
|
cls.borrower1 = Borrower.objects.create(name="Jenny")
|
|
cls.borrower2 = Borrower.objects.create(name="Kevin")
|
|
# borrower 1 reserves, rents, and returns book1.
|
|
Reservation.objects.create(
|
|
borrower=cls.borrower1,
|
|
book=cls.book1,
|
|
state=Reservation.STOPPED,
|
|
)
|
|
RentalSession.objects.create(
|
|
borrower=cls.borrower1,
|
|
book=cls.book1,
|
|
state=RentalSession.STOPPED,
|
|
)
|
|
# borrower2 reserves, rents, and returns book1.
|
|
Reservation.objects.create(
|
|
borrower=cls.borrower2,
|
|
book=cls.book1,
|
|
state=Reservation.STOPPED,
|
|
)
|
|
RentalSession.objects.create(
|
|
borrower=cls.borrower2,
|
|
book=cls.book1,
|
|
state=RentalSession.STOPPED,
|
|
)
|
|
|
|
def test_aggregate(self):
|
|
"""
|
|
filtered_relation() not only improves performance but also creates
|
|
correct results when aggregating with multiple LEFT JOINs.
|
|
|
|
Books can be reserved then rented by a borrower. Each reservation and
|
|
rental session are recorded with Reservation and RentalSession models.
|
|
Every time a reservation or a rental session is over, their state is
|
|
changed to 'stopped'.
|
|
|
|
Goal: Count number of books that are either currently reserved or
|
|
rented by borrower1 or available.
|
|
"""
|
|
qs = (
|
|
Book.objects.annotate(
|
|
is_reserved_or_rented_by=Case(
|
|
When(
|
|
reservation__state=Reservation.NEW,
|
|
then=F("reservation__borrower__pk"),
|
|
),
|
|
When(
|
|
rental_session__state=RentalSession.NEW,
|
|
then=F("rental_session__borrower__pk"),
|
|
),
|
|
default=None,
|
|
)
|
|
)
|
|
.filter(
|
|
Q(is_reserved_or_rented_by=self.borrower1.pk) | Q(state=Book.AVAILABLE)
|
|
)
|
|
.distinct()
|
|
)
|
|
self.assertEqual(qs.count(), 1)
|
|
# If count is equal to 1, the same aggregation should return in the
|
|
# same result but it returns 4.
|
|
self.assertSequenceEqual(
|
|
qs.annotate(total=Count("pk")).values("total"), [{"total": 4}]
|
|
)
|
|
# With FilteredRelation, the result is as expected (1).
|
|
qs = (
|
|
Book.objects.annotate(
|
|
active_reservations=FilteredRelation(
|
|
"reservation",
|
|
condition=Q(
|
|
reservation__state=Reservation.NEW,
|
|
reservation__borrower=self.borrower1,
|
|
),
|
|
),
|
|
)
|
|
.annotate(
|
|
active_rental_sessions=FilteredRelation(
|
|
"rental_session",
|
|
condition=Q(
|
|
rental_session__state=RentalSession.NEW,
|
|
rental_session__borrower=self.borrower1,
|
|
),
|
|
),
|
|
)
|
|
.filter(
|
|
(
|
|
Q(active_reservations__isnull=False)
|
|
| Q(active_rental_sessions__isnull=False)
|
|
)
|
|
| Q(state=Book.AVAILABLE)
|
|
)
|
|
.distinct()
|
|
)
|
|
self.assertEqual(qs.count(), 1)
|
|
self.assertSequenceEqual(
|
|
qs.annotate(total=Count("pk")).values("total"), [{"total": 1}]
|
|
)
|
|
|
|
def test_condition_spans_join(self):
|
|
self.assertSequenceEqual(
|
|
Book.objects.annotate(
|
|
contains_editor_author=FilteredRelation(
|
|
"author", condition=Q(author__name__icontains=F("editor__name"))
|
|
)
|
|
).filter(
|
|
contains_editor_author__isnull=False,
|
|
),
|
|
[self.book1],
|
|
)
|
|
|
|
def test_condition_spans_join_chained(self):
|
|
self.assertSequenceEqual(
|
|
Book.objects.annotate(
|
|
contains_editor_author=FilteredRelation(
|
|
"author", condition=Q(author__name__icontains=F("editor__name"))
|
|
),
|
|
contains_editor_author_ref=FilteredRelation(
|
|
"author",
|
|
condition=Q(author__name=F("contains_editor_author__name")),
|
|
),
|
|
).filter(
|
|
contains_editor_author_ref__isnull=False,
|
|
),
|
|
[self.book1],
|
|
)
|
|
|
|
def test_condition_self_ref(self):
|
|
self.assertSequenceEqual(
|
|
Book.objects.annotate(
|
|
contains_author=FilteredRelation(
|
|
"author",
|
|
condition=Q(title__icontains=F("author__name")),
|
|
)
|
|
).filter(
|
|
contains_author__isnull=False,
|
|
),
|
|
[self.book1],
|
|
)
|
|
|
|
|
|
class FilteredRelationAnalyticalAggregationTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
author = Author.objects.create(name="Author")
|
|
editor = Editor.objects.create(name="Editor")
|
|
cls.book1 = Book.objects.create(
|
|
title="Poem by Alice",
|
|
editor=editor,
|
|
author=author,
|
|
)
|
|
cls.book2 = Book.objects.create(
|
|
title="The book by Jane A",
|
|
editor=editor,
|
|
author=author,
|
|
)
|
|
cls.book3 = Book.objects.create(
|
|
title="The book by Jane B",
|
|
editor=editor,
|
|
author=author,
|
|
)
|
|
cls.seller1 = Seller.objects.create(name="Seller 1")
|
|
cls.seller2 = Seller.objects.create(name="Seller 2")
|
|
cls.usd = Currency.objects.create(currency="USD")
|
|
cls.eur = Currency.objects.create(currency="EUR")
|
|
cls.sales_date1 = date(2020, 7, 6)
|
|
cls.sales_date2 = date(2020, 7, 7)
|
|
ExchangeRate.objects.bulk_create(
|
|
[
|
|
ExchangeRate(
|
|
rate_date=cls.sales_date1,
|
|
from_currency=cls.usd,
|
|
to_currency=cls.eur,
|
|
rate=0.40,
|
|
),
|
|
ExchangeRate(
|
|
rate_date=cls.sales_date1,
|
|
from_currency=cls.eur,
|
|
to_currency=cls.usd,
|
|
rate=1.60,
|
|
),
|
|
ExchangeRate(
|
|
rate_date=cls.sales_date2,
|
|
from_currency=cls.usd,
|
|
to_currency=cls.eur,
|
|
rate=0.50,
|
|
),
|
|
ExchangeRate(
|
|
rate_date=cls.sales_date2,
|
|
from_currency=cls.eur,
|
|
to_currency=cls.usd,
|
|
rate=1.50,
|
|
),
|
|
ExchangeRate(
|
|
rate_date=cls.sales_date2,
|
|
from_currency=cls.usd,
|
|
to_currency=cls.usd,
|
|
rate=1.00,
|
|
),
|
|
]
|
|
)
|
|
BookDailySales.objects.bulk_create(
|
|
[
|
|
BookDailySales(
|
|
book=cls.book1,
|
|
sale_date=cls.sales_date1,
|
|
currency=cls.usd,
|
|
sales=100.00,
|
|
seller=cls.seller1,
|
|
),
|
|
BookDailySales(
|
|
book=cls.book2,
|
|
sale_date=cls.sales_date1,
|
|
currency=cls.eur,
|
|
sales=200.00,
|
|
seller=cls.seller1,
|
|
),
|
|
BookDailySales(
|
|
book=cls.book1,
|
|
sale_date=cls.sales_date2,
|
|
currency=cls.usd,
|
|
sales=50.00,
|
|
seller=cls.seller2,
|
|
),
|
|
BookDailySales(
|
|
book=cls.book2,
|
|
sale_date=cls.sales_date2,
|
|
currency=cls.eur,
|
|
sales=100.00,
|
|
seller=cls.seller2,
|
|
),
|
|
]
|
|
)
|
|
|
|
def test_aggregate(self):
|
|
tests = [
|
|
Q(daily_sales__sale_date__gte=self.sales_date2),
|
|
~Q(daily_sales__seller=self.seller1),
|
|
]
|
|
for condition in tests:
|
|
with self.subTest(condition=condition):
|
|
qs = (
|
|
Book.objects.annotate(
|
|
recent_sales=FilteredRelation(
|
|
"daily_sales", condition=condition
|
|
),
|
|
recent_sales_rates=FilteredRelation(
|
|
"recent_sales__currency__rates_from",
|
|
condition=Q(
|
|
recent_sales__currency__rates_from__rate_date=F(
|
|
"recent_sales__sale_date"
|
|
),
|
|
recent_sales__currency__rates_from__to_currency=(
|
|
self.usd
|
|
),
|
|
),
|
|
),
|
|
)
|
|
.annotate(
|
|
sales_sum=Sum(
|
|
F("recent_sales__sales") * F("recent_sales_rates__rate"),
|
|
output_field=DecimalField(),
|
|
),
|
|
)
|
|
.values("title", "sales_sum")
|
|
.order_by(
|
|
F("sales_sum").desc(nulls_last=True),
|
|
)
|
|
)
|
|
self.assertSequenceEqual(
|
|
qs,
|
|
[
|
|
{"title": self.book2.title, "sales_sum": Decimal(150.00)},
|
|
{"title": self.book1.title, "sales_sum": Decimal(50.00)},
|
|
{"title": self.book3.title, "sales_sum": None},
|
|
],
|
|
)
|