在django-fsm中配置外键类型FSM字段失败:状态存为字符串
问题描述
希望将Task模型的工作流状态字段设置为WorkFlowState模型的外键,同时通过Transition模型定义FSM状态转换规则,但配置后task_state字段被存储为字符串,而非预期的外键关联实例。
原代码
import datetime from django.db import models from django_fsm import FSMField, transition class WorkFlowState(models.Model): name = models.CharField(max_length=64, unique=True) preceding = models.ForeignKey( 'self', on_delete=models.CASCADE, blank=True, null=True) # postAction = models.CharField(max_length=64, null=True, blank=True) default_warn1 = models.DurationField( blank=False, null=False, default=datetime.timedelta(hours=12)) default_warn2 = models.DurationField( blank=False, null=False, default=datetime.timedelta(days=1)) default_duration = models.DurationField( blank=False, null=False, default=datetime.timedelta(days=1)) def __str__(self): return self.name class Transition(models.Model): name = models.CharField(max_length=64) source_state = models.ForeignKey(WorkFlowState, on_delete=models.CASCADE, related_name='transitions_from') target_state = models.ForeignKey(WorkFlowState, on_delete=models.CASCADE, related_name='transitions_to') def __str__(self): return self.name class Task(models.Model): name = models.CharField(max_length=100) assigned_on = models.DateTimeField(auto_now=True) created_on = models.DateTimeField(auto_now_add=True) task_state = FSMField(models.ForeignKey(WorkFlowState, on_delete=models.PROTECT), protected=True) def __str__(self): return self.task_state def perform_transition(self, transition_name): try: transition = Transition.objects.get(name=transition_name, source_state=self.task_state) setattr(self, 'assigned_to', transition.target_state) self.save() return True except Transition.DoesNotExist: return False
问题分析与修复方案
核心问题
- FSMField 使用错误:django_fsm的
FSMField本质是字符串类型字段,不能直接将ForeignKey作为参数传入。原代码中错误地将外键字段类型传给FSMField,导致字段存储的是外键实例的字符串表示,而非外键关联。 - 状态转换字段名错误:
perform_transition方法中错误地将状态赋值给assigned_to字段,实际应更新的是task_state。 - __str__方法类型错误:直接返回外键实例会触发类型错误,需转为字符串。
修复方案一:直接使用ForeignKey配合django_fsm
适合需要数据库外键约束、直接关联WorkFlowState实例的场景:
import datetime from django.db import models from django_fsm import transition class WorkFlowState(models.Model): name = models.CharField(max_length=64, unique=True) preceding = models.ForeignKey( 'self', on_delete=models.CASCADE, blank=True, null=True) default_warn1 = models.DurationField( blank=False, null=False, default=datetime.timedelta(hours=12)) default_warn2 = models.DurationField( blank=False, null=False, default=datetime.timedelta(days=1)) default_duration = models.DurationField( blank=False, null=False, default=datetime.timedelta(days=1)) def __str__(self): return self.name class Transition(models.Model): name = models.CharField(max_length=64) source_state = models.ForeignKey(WorkFlowState, on_delete=models.CASCADE, related_name='transitions_from') target_state = models.ForeignKey(WorkFlowState, on_delete=models.CASCADE, related_name='transitions_to') def __str__(self): return self.name class Task(models.Model): name = models.CharField(max_length=100) assigned_on = models.DateTimeField(auto_now=True) created_on = models.DateTimeField(auto_now_add=True) # 改为ForeignKey,需提前创建默认的WorkFlowState实例并指定default task_state = models.ForeignKey(WorkFlowState, on_delete=models.PROTECT) def __str__(self): return f"{self.name} - {str(self.task_state)}" @transition(field='task_state', source='*', target=None) def perform_transition(self, transition_name): try: transition_obj = Transition.objects.get(name=transition_name, source_state=self.task_state) self.task_state = transition_obj.target_state self.save() return True except Transition.DoesNotExist: return False
修复方案二:FSMField存储状态标识关联实例
适合需要利用django_fsm状态保护、转换历史等功能的场景:
import datetime from django.db import models from django_fsm import FSMField class WorkFlowState(models.Model): name = models.CharField(max_length=64, unique=True) preceding = models.ForeignKey( 'self', on_delete=models.CASCADE, blank=True, null=True) default_warn1 = models.DurationField( blank=False, null=False, default=datetime.timedelta(hours=12)) default_warn2 = models.DurationField( blank=False, null=False, default=datetime.timedelta(days=1)) default_duration = models.DurationField( blank=False, null=False, default=datetime.timedelta(days=1)) def __str__(self): return self.name class Transition(models.Model): name = models.CharField(max_length=64) source_state = models.ForeignKey(WorkFlowState, on_delete=models.CASCADE, related_name='transitions_from') target_state = models.ForeignKey(WorkFlowState, on_delete=models.CASCADE, related_name='transitions_to') def __str__(self): return self.name class Task(models.Model): name = models.CharField(max_length=100) assigned_on = models.DateTimeField(auto_now=True) created_on = models.DateTimeField(auto_now_add=True) # 用FSMField存储状态名称(依赖WorkFlowState的name唯一约束) task_state_name = FSMField(max_length=64, protected=True) @property def task_state(self): return WorkFlowState.objects.get(name=self.task_state_name) @task_state.setter def task_state(self, state): self.task_state_name = state.name def __str__(self): return f"{self.name} - {self.task_state_name}" def perform_transition(self, transition_name): try: transition_obj = Transition.objects.get(name=transition_name, source_state=self.task_state) self.task_state = transition_obj.target_state self.save() return True except (Transition.DoesNotExist, WorkFlowState.DoesNotExist): return False
内容的提问来源于stack exchange,提问作者Reza Teymoori
相关产品推荐
相关产品推荐

