Is it possible to block execution of a trigger if the transition isn't valid?

Question

It appears the trigger methods still run then raise the MachineError exception afterwards when transition is not valid from current state. Is there a way to block the execution of a trigger so that a call to the trigger on the model will simply raise the exception and not execute the trigger?

Sorry, forgot to mention using the overridden _checked_assignment from the FAQ which may be reason for this behavior.

from transitions import State, Machine

class StateMachine(Machine):

    def _checked_assignment(self, model, name, func):
        if hasattr(model, name):
            predefined_func = getattr(model, name)
            def nested_func(*args, **kwargs):
                predefined_func()
                func(*args, **kwargs)
            setattr(model, name, nested_func)
        else:
            setattr(model, name, func)

class Rocket(StateMachine):
    def __init__():
        StateMachine.__init__(
            self,
            states=["on_pad", "fueling", "ready", "launched", "meco", "second_stage", "orbit"],
            transitions=[
                {'trigger': 'fuel', 'source': 'on_pad', 'dest': 'fueling'},
                {'trigger': 'power_on', 'source': 'fueling', 'dest': 'ready'},
                {'trigger': 'launch', 'source': 'ready', 'dest': 'launched'}
            ],
            initial='on_pad'
        )

    def fuel():
        print("cryos loading...")

    def launch():
        print("launching")

def main():
    rocket = Rocket()
    rocket.launch()  # prints "launching" then throws Machine Error, need to block actual method execution

Could you provide a code example for your issue? The 'trigger' includes the check whether a transition is valid or not from the current state but will not process the actual transition. — aleneum, Sep 15 '21 at 06:56
Added example. Apologies, forgot to mention using the overriden `_checked_assignment` from FAQ which I believe is the reason for the behavior — Josh Alexander, Sep 15 '21 at 17:27

score 0 · Answer 1 · answered Sep 15 '21 at 19:24

Found the issue as I typed out the example, need to call the passed in func() prior to predefined_func() within _checked_assignment. This method should be updated on the FAQ to retain original transition functionality.

    def _checked_assignment(self, model, name, func):
        if hasattr(model, name):
            predefined_func = getattr(model, name)
            def nested_func(*args, **kwargs):
                func(*args, **kwargs)  # need to call before pre-defined
                predefined_func()
            setattr(model, name, nested_func)
        else:
            setattr(model, name, func)

aleneum · Accepted Answer · 2021-09-16T07:08:51.020

While you could wrap your callbacks with an override of Machine._checked_assignment as described in this answer, I'd recommend tying methods that should be called in the context of a transition to its callbacks. Callbacks can be called on multiple occasions during a transition as described in the documentation's chapter Callback execution order. The caveat is that callbacks must not have the same name as intended triggers but this is usually a minor setback and also enables you to add multiple callbacks to the same event. I reworked your example a bit. Rocket acts as the stateful model but the machine itself has been separated. You could also manage the state machine completely independently of your Rocket in case you plan to use multiple instances. One machine can handle multiple stateful objects. Furthermore, I renamed your callbacks slightly and passed them to the before keyword of the transitions. As mentioned earlier, this could also be a list ({'before': ['on_launch']} is also valid). This way, they will be called right before the transition will happen and will not be called when a) Rocket is not in the correct state or b) condition checks for the transition in question failed.

from transitions import Machine, MachineError


class Rocket:
    
    def __init__(self):
        self.machine = Machine(
            self,
            states=["on_pad", "fueling", "ready", "launched", "meco", "second_stage", "orbit"],
            transitions=[
                {'trigger': 'fuel', 'source': 'on_pad', 'dest': 'fueling', 'before': 'on_fueling'},
                {'trigger': 'power_on', 'source': 'fueling', 'dest': 'ready'},
                {'trigger': 'launch', 'source': 'ready', 'dest': 'launched', 'before': 'on_launch'}
            ],
            initial='on_pad'
        )

    def on_fueling(self):
        print("cryos loading...")

    def on_launch(self):
        print("launching")


rocket = Rocket()
try:
    rocket.launch()
    assert False
except MachineError:
    pass

rocket.fuel()  # >>> cryos loading...
rocket.power_on() 
rocket.launch()  # >>> launching

Is it possible to block execution of a trigger if the transition isn't valid?

2 Answers2