You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Django信号处理器无效化cached_property的实现是否正确?有无边缘情况?

Django缓存无效化问题:Team.total_points的信号处理器是否覆盖所有场景?

模型定义

假设我有以下Django模型:

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()

需求

我希望创建一个信号处理器,在TeamUser对象创建/更新/删除时,无效化team.total_points的缓存。

初始实现

我最初编写了如下信号处理器,按照Django文档推荐的del instance.prop方式来无效化缓存:

@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

测试用例(基于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

遇到的问题

test_create_team_users测试通过,但test_delete_all_team_users测试失败:实际查询数为12而非预期的2,出现了N+1查询问题。这是因为批量删除时,每个TeamUser触发信号都会去数据库获取对应的Team实例,导致额外的查询开销。

改进后的实现

为解决该问题,我更新了信号处理器,仅当TeamUser对象关联的team已在实例中缓存时,才去无效化team.total_points的缓存(is_cached方法用于判断外键是否已缓存):

@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

现在两个测试都通过了!

疑问

请问该实现能否在所有场景下正确无效化team.total_points的缓存?是否存在遗漏的边缘情况?


内容的提问来源于stack exchange,提问作者Johnny Metz

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.12 02:33:34