1. Ian George
  2. django-statemachine

Commits

Ian George  committed 850acca

complete rewrite of declaration code, now django-style declarative, tests passing but needs more attention

  • Participants
  • Parent commits f77d006
  • Branches default

Comments (0)

Files changed (3)

File statemachine/fields.py

View file
 from copy import deepcopy
 from django.db import models
-from fsm import FSM_State, FSM
+from fsm import State, FSM
 from forms import FSM_StateFormField
 
 DEFAULT_STATE = "start"
 
 class FSM_StateField(models.Field):
     __metaclass__ = models.SubfieldBase
-    _machine = None
 
     def __init__(self, machine, *args, **kwargs):
-        """
-        machine must be a class, not an instance
-        """
-        self.__base_machine = machine
+        self.__machine_class = machine
         defaults = {'max_length': 50, 'default' : DEFAULT_STATE}
         defaults.update(kwargs)
         super(FSM_StateField, self).__init__(self, **defaults)
 
-    def setup(self, state_name):
-        if not self._machine:
-            self._machine = deepcopy(self.__base_machine)
-
-        self._machine.set_initial_state(state_name)
 
     def db_type(self, connection):
         return "char(50)"
 
     def to_python(self, value):
+        self.machine = self.__machine_class()
         try:
             name = value.state
         except AttributeError:
             name = value
 
-        self.setup(name)
+        self.machine.set_initial_state(name)
 
-        return self._machine
+
+        return self.machine
 
     def get_prep_value(self, value):
         return value
             name = value
         return name
 
+
     def formfield(self, **kwargs):
-        if self._machine:
-            choices = tuple(self._machine.get_django_choices())
+        if self.machine:
+            choices = tuple(self.machine.get_django_choices())
         else:
             choices = ((DEFAULT_STATE, DEFAULT_STATE),)
         defaults = {'form_class': FSM_StateFormField, 'label':self.name, 'choices':choices}

File statemachine/fsm.py

View file
+from copy import deepcopy
 from django.conf import settings
-
 """
 Some Exceptions
 """
     STATE_MACHINE_DEFAULT = "start"
 
 
-class FSM_State(object):
+from django.utils.datastructures import SortedDict
+
+def get_declared_states(bases, attrs, with_base_states=True):
+    """
+    ** "Inspired by" django.forms.forms **
+
+    Create a list of State instances from the passed in 'attrs', plus any
+    similar states on the base classes (in 'bases').
+
+    If 'with_base_states' is True, all states from the bases are used.
+    Otherwise, only states in the 'declared_states' attribute on the bases are
+    used.
+    """
+    states = [(state_name, attrs.pop(state_name)) for state_name, obj in attrs.items() if isinstance(obj, State)]
+    states.sort(key=lambda x: x[1].creation_counter)
+
+    # If this class is subclassing another StateMachine, add that StateMachine's states.
+    # Note that we loop over the bases in *reverse*. This is necessary in
+    # order to preserve the correct order of statess.
+    if with_base_states:
+        for base in bases[::-1]:
+            if hasattr(base, 'base_states'):
+                states = base.base_states.items() + states
+    else:
+        for base in bases[::-1]:
+            if hasattr(base, 'declared_states'):
+                states = base.declared_states.items() + states
+
+    return SortedDict(states)
+
+class DeclarativeStatesMetaclass(type):
+    """
+    Metaclass that converts State attributes to a dictionary called
+    'base_states', taking into account parent class 'base_states' as well.
+    """
+    def __new__(cls, name, bases, attrs):
+        attrs['base_states'] = get_declared_states(bases, attrs)
+        new_class = super(DeclarativeStatesMetaclass,
+                     cls).__new__(cls, name, bases, attrs)
+        return new_class
+
+class State(object):
     """
     Represents each individual state of the machine. 
 
     exit_states = []
     entry_action = None
     exit_action = None
+    creation_counter = 0
 
-    def __init__(self, name, exit_states, entry_action=None, exit_action=None):
-        self.name = name
-        self.exit_states = exit_states
+    def __init__(self, label=None, exits_to=None, entry_action=None, exit_action=None):
+        self.label = label
+        self.exit_states = exits_to or []
         self.entry_action = entry_action
         self.exit_action = exit_action
 
-
     def exit(self, target_state, *args, **kwargs):
         """ 
         Checks states and exits if possible, if not possible it raises FSM_TransitionNotAllowed 
     """
     Simple FSM implementation.
     """
-    def __init__(self, initial_state, verify_on_execute=True):
-        self.states = {}
-        self.add(initial_state)
-        self.__state = initial_state.name
+    __metaclass__ = DeclarativeStatesMetaclass
+
+    def __init__(self, verify_on_execute=True):
+        self.states = deepcopy(self.base_states)
+        self.__state = self.states[self.states.keys()[0]].name
         self.dbg = None
         self.verify_on_execute = verify_on_execute
 
         
     def verify(self):
         """
-        Best only called on fully formed machines, checks for exit states that don't exist
         """
         bad_states = []
         state_names = set(self.states.keys())
 
         #TODO: this needs to be transactional, shouldn't exit unless it
         #can enter
-
         exiting_state.exit(entering_state, *args, **kwargs)
         entering_state.enter(entering_state, *args, **kwargs)
 

File statemachine/tests.py

View file
 from django.test.client import Client
 
 from models import StateMachineWithHistory
-from fsm import FSM, FSM_State, FSM_NotAllowed, FSM_TransitionNotAllowed, FSM_VerificationError
 from fields import FSM_StateField
+import fsm
 
-testfsm = FSM(FSM_State("start" ,["step1a", "step1b", "step1c"]))
-testfsm.add(FSM_State("step1a", ["step2",]))
-testfsm.add(FSM_State("step1b", ["step2",]))
-testfsm.add(FSM_State("step1c", ["step2",]))
-testfsm.add(FSM_State("step2", ["end", "step3"]))
-testfsm.add(FSM_State("step3", ["end",]))
-testfsm.add(FSM_State("end", []))
+class TestMachine(fsm.FSM):
+    start = fsm.State(exits_to=['step1a', 'step1b', 'step1c'])
+    step1a = fsm.State(exits_to=['step2',])
+    step1b = fsm.State(exits_to=['step2',])
+    step1c = fsm.State(exits_to=['step2',])
+    step2 = fsm.State(exits_to=['end', 'step3'])
+    step3 = fsm.State(exits_to=['end',])
+    end = fsm.State()
 
 class TestStateMachine(StateMachineWithHistory):
-    status = FSM_StateField(testfsm)
+    status = FSM_StateField(TestMachine)
     field1 = models.IntegerField()
     field2 = models.CharField(max_length=25)
 
     def test_verify(self):
         """ Should raise a FSM_VerificationError """
         testmachine = TestStateMachine(field1=100, field2="LALALALALA")
-        testmachine.status.add(FSM_State("random", ["doesntexist", "alsodoesnexist"]))
-        with self.assertRaises(FSM_VerificationError):
+        testmachine.status.add(fsm.State("random", ["doesntexist", "alsodoesnexist"]))
+        with self.assertRaises(fsm.FSM_VerificationError):
             testmachine.status.verify()
 
     def test_save_fail(self):
         """Should raise a StateMachineNotAllowed, messing with the state outside of the machine."""
         testmachine = TestStateMachine(field1=100, field2="LALALALALA")
         testmachine.save(None)
-        with self.assertRaises(FSM_NotAllowed):
+        with self.assertRaises(fsm.FSM_NotAllowed):
             testmachine.status.state = "Testing"    
             testmachine.save(None)
 
         #check we are starting at "start"
         self.assertEqual(testmachine.status.state, "start")
         
-        with self.assertRaises(FSM_TransitionNotAllowed):
+        with self.assertRaises(fsm.FSM_TransitionNotAllowed):
             testmachine.status.change("step3")
 
         #check we are still at "start" after a failed transition
         """ Tests an action set to run on entry to a state """
         def entry_func(entered, row):
             row.field1 = 500
-
         testmachine = TestStateMachine(field1=100, field2="LALALALALA")
         testmachine.save(None)
         testmachine.status.states["step3"].entry_action = entry_func
         testmachine.save(None)
         testmachine.status.change("step3", testmachine)
         testmachine.save(None)
-        
         self.assertEqual(testmachine.field1, 500)
 
     def test_exit_action(self):