如何基于Django FSM实现父模型状态随子模型状态自动变更

嘿，这个需求我之前做过类似的，用Django FSM实现起来思路很清晰，我给你一步步拆解具体实现方案：

核心思路 #

当任意一个Bar的状态切换到complete时，自动检查它关联的Foo下所有Bar是否都处于complete状态，如果是，就触发Foo的状态转换到complete。我们可以通过Django FSM的状态转换钩子+事务来实现这个逻辑。

第一步：完善模型的状态定义与转换规则 #

首先给两个模型补充完整的状态选项和合法的状态转换路径：

from django.db import models
from django_fsm import FSMField, transition
from django.db import transaction

class Foo(models.Model):
    baz = models.CharField(max_length=100)
    # 定义Foo的所有状态选项
    state = FSMField(
        default='new',
        choices=[
            ('new', 'New'),
            ('draft', 'Draft'),
            ('complete', 'Complete'),
            ('pending_approval', 'Pending Approval'),
            ('approved', 'Approved'),
        ]
    )

    def all_associated_bars_complete(self):
        """辅助方法：检查当前Foo下所有关联的Bar是否都为complete状态"""
        # 使用select_for_update加锁，避免并发修改导致的竞态问题
        return not self.bar_set.select_for_update().filter(state__ne='complete').exists()

    @transition(
        field=state,
        source=['new', 'draft', 'pending_approval'],  # 允许从这些状态转到complete
        target='complete'
    )
    def transition_to_complete(self):
        """Foo状态转换到complete的方法，可在这里添加额外逻辑（如日志）"""
        pass

class Bar(models.Model):
    x = models.CharField(max_length=100)
    # 定义Bar的状态选项
    state = FSMField(
        default='draft',
        choices=[
            ('draft', 'Draft'),
            ('complete', 'Complete'),
        ]
    )
    foo = models.ForeignKey(Foo, on_delete=models.CASCADE)

    @transition(
        field=state,
        source='draft',
        target='complete'
    )
    @transaction.atomic  # 用事务确保Bar状态更新和Foo状态转换的原子性
    def mark_as_complete(self):
        """Bar从draft转换到complete的方法，触发Foo的状态检查"""
        # 先保存当前Bar的状态变更
        self.save()
        
        # 锁定关联的Foo对象，防止并发修改
        foo = Foo.objects.select_for_update().get(pk=self.foo.pk)
        
        # 检查所有关联Bar是否都完成
        if foo.all_associated_bars_complete():
            foo.transition_to_complete()
            foo.save()

关键细节说明 #

1. 状态转换的合法性：通过@transition装饰器明确定义每个状态能转换到的目标状态，符合Django FSM的规范，避免非法状态变更。
2. 并发问题处理：使用select_for_update锁定Foo对象，加上@transaction.atomic事务，防止多个Bar同时更新时出现竞态条件（比如两个请求都检查到还有一个Bar未完成，最终Foo没被标记为complete）。
3. 状态检查的准确性：每次检查前重新查询并锁定Foo，确保拿到最新的关联Bar状态，避免缓存导致的判断错误。

额外扩展（可选） #

如果业务需要支持Bar从complete转回draft时，自动将Foo的状态从complete回退（比如回到draft），可以给Bar添加反向转换的方法，类似逻辑：

@transition(
    field=state,
    source='complete',
    target='draft'
)
@transaction.atomic
def mark_as_draft(self):
    self.save()
    foo = Foo.objects.select_for_update().get(pk=self.foo.pk)
    # 如果Foo当前是complete状态，且现在有Bar不是complete了，就转回draft
    if foo.state == 'complete' and not foo.all_associated_bars_complete():
        # 记得给Foo定义从complete转回draft的transition
        foo.transition_to_draft()
        foo.save()

内容的提问来源于stack exchange，提问作者at14