Let's say I have the following Django models:
class Team(models.Model):
users = models.ManyToManyField(User, through="TeamUser")
@cached_property
def total_points(self):
return self.teamuser_set.aggregate(models.Sum("points"))["points__sum"] or 0
class TeamUser(models.Model):
team = models.ForeignKey(Team, on_delete=models.CASCADE)
user = models.ForeignKey(User, on_delete=models.CASCADE)
points = models.IntegerField()
I want to create a signal handler that will invalidate the team.total_points
cache when TeamUser
object is created/updated/deleted.
I started with the following signal handler. Note the Django docs recommended the del instance.prop
call.
@receiver(post_save, sender=models.TeamUser)
@receiver(post_delete, sender=models.TeamUser)
def invalidate_cache(**kwargs):
try:
del kwargs["instance"].team.total_points
except AttributeError:
pass
And some tests. Note I'm using pytest-django.
def test_create_team_users(django_assert_num_queries):
user = factories.UserFactory()
team = factories.TeamFactory()
assert team.total_points == 0
with django_assert_num_queries(1):
TeamUser.objects.create(team=team, user=user, points=2)
assert team.total_points == 2
with django_assert_num_queries(1):
TeamUser.objects.create(team=team, user=user, points=3)
assert team.total_points == 5
def test_delete_all_team_users(django_assert_num_queries):
user = factories.UserFactory()
team = factories.TeamFactory()
for _ in range(10):
TeamUser.objects.create(team=team, user=user, points=2)
with django_assert_num_queries(2):
TeamUser.objects.all().delete()
assert team.total_points == 0
The test_create_team_users
test passed but the test_delete_all_team_users
test failed because the query count is 12 instead of 2. Yikes! Looks like an N+1 query.
To prevent this, I updated my signal handler to only invalidate the team.total_points
cache if the user object is cached on the TeamUser
object. I found the is_cached
method in this SO answer.
@receiver(post_save, sender=models.TeamUser)
@receiver(post_delete, sender=models.TeamUser)
def invalidate_cache(sender, instance, **kwargs):
if sender.team.is_cached(instance):
try:
del instance.team.total_points
except AttributeError:
pass
Now both tests pass!
Does this correctly invalidate the team.total_points
cache in all cases? Is there an edge case I'm missing?
from Invalidate Django cached_property in signal handler without introducing unnecessary queries
No comments:
Post a Comment