-
-
Notifications
You must be signed in to change notification settings - Fork 105
/
transition.py
70 lines (54 loc) · 2.43 KB
/
transition.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import logging
from django.db.models import CASCADE, PROTECT
from river.models import State, Workflow, TransitionMeta
try:
from django.contrib.contenttypes.fields import GenericForeignKey
except ImportError:
from django.contrib.contenttypes.generic import GenericForeignKey
from django.db import models
from django.utils.translation import ugettext_lazy as _
from river.models.base_model import BaseModel
from river.models.managers.transitionapproval import TransitionApprovalManager
from river.config import app_config
PENDING = "pending"
CANCELLED = "cancelled"
JUMPED = "jumped"
DONE = "done"
STATUSES = [
(PENDING, _('Pending')),
(CANCELLED, _('Cancelled')),
(DONE, _('Done')),
(JUMPED, _('Jumped')),
]
LOGGER = logging.getLogger(__name__)
class Transition(BaseModel):
class Meta:
app_label = 'river'
verbose_name = _("Transition")
verbose_name_plural = _("Transitions")
objects = TransitionApprovalManager()
content_type = models.ForeignKey(app_config.CONTENT_TYPE_CLASS, verbose_name=_('Content Type'), on_delete=CASCADE)
object_id = models.CharField(max_length=50, verbose_name=_('Related Object'))
workflow_object = GenericForeignKey('content_type', 'object_id')
meta = models.ForeignKey(TransitionMeta, verbose_name=_('Meta'), related_name="transitions", on_delete=PROTECT)
workflow = models.ForeignKey(Workflow, verbose_name=_("Workflow"), related_name='transitions', on_delete=PROTECT)
source_state = models.ForeignKey(State, verbose_name=_("Source State"), related_name='transition_as_source', on_delete=PROTECT)
destination_state = models.ForeignKey(State, verbose_name=_("Destination State"), related_name='transition_as_destination', on_delete=PROTECT)
status = models.CharField(_('Status'), choices=STATUSES, max_length=100, default=PENDING)
iteration = models.IntegerField(default=0, verbose_name=_('Priority'))
@property
def next_transitions(self):
return Transition.objects.filter(
workflow=self.workflow,
workflow_object=self.workflow_object,
source_state=self.destination_state,
iteration=self.iteration + 1
)
@property
def peers(self):
return Transition.objects.filter(
workflow=self.workflow,
workflow_object=self.workflow_object,
source_state=self.source_state,
iteration=self.iteration
).exclude(pk=self.pk)