Can we have an example persisting the state machine with sqlalchemy or redis/valkey

[shun-liang]

Dear python-statemachine developers

Many thanks for this library.

I am very keen to adapt this library for my project, but I am struggling thinking how I can persist the state machine with redis or valkey.

I have found this example in the doc, but it's not very clear to me what's going on there - is FilePersistentModel actually persisting the bytes representation of the state?

I have also found the docs about domain models, and Django integration example that is based on the Mixin which looks very convenient. I don't use Django, but is it right ot assume this can be transferred quite easily to sqlalchemy? If so, would you mind adding some support or an example in the docs?

Also wonder if there is an easy way to persist the state with redis or valkey.

Many thanks again.

[fgmacedo]

Hi @shun-liang , how are you? Thanks for getting in touch.

Do you need any extra information to be persisted or only the current state?

[fgmacedo]

For example, a quick way to only persist the current state is to implement the on_enter_state hook:

Redis persistence state machine using callback hook

Demonstrate a simple way of how to use redis to persist the current state.

This example uses the start_value constructor param and the on_enter_state callback hook.

from statemachine import State
from statemachine import StateMachine

try:
    # We try to import real redis client
    # if it fails we use a fake one only for demostration
    from redis import StrictRedis
except ImportError:

    class StrictRedis:
        """Fake redis.StrictRedis client only for demostration purposes"""

        def __init__(self, *args, **kwargs):
            self._fake_state = {}

        def set(self, key, value):
            self._fake_state[key] = value

        def get(self, key):
            return self._fake_state.get(key)


redis_client = StrictRedis(host="localhost", port=6379, decode_responses=True)


class TrafficLightMachine(StateMachine):
    "A traffic light machine"

    green = State(initial=True)
    yellow = State()
    red = State()

    cycle = green.to(yellow) | yellow.to(red) | red.to(green)

    def __init__(self, *args, redis_key: str, **kwargs):
        self.redis_key = redis_key
        start_value = redis_client.get(redis_key)
        super().__init__(*args, start_value=start_value, **kwargs)

    def on_enter_state(self, state: State):
        redis_client.set(self.redis_key, state.id)

Executing

key = "traffic_light"
sm = TrafficLightMachine(redis_key=key)
sm.send("cycle")

assert sm.current_state.id == "yellow"

del sm

sm = TrafficLightMachine(redis_key=key)

assert sm.current_state.id == "yellow"

[fgmacedo]

Another example using the model method. The final result is the same of the previous example using custom hooks.

Redis persistence state machine using a custom model

Demonstrate a simple way to use redis to persist the current state using a custom model.

from statemachine import State
from statemachine import StateMachine

try:
    # We try to import real redis client
    # if it fails we use a fake one only for demostration
    from redis import StrictRedis
except ImportError:

    class StrictRedis:
        """Fake redis.StrictRedis client only for demostration purposes"""

        def __init__(self, *args, **kwargs):
            self._fake_state = {}

        def set(self, key, value):
            self._fake_state[key] = value

        def get(self, key):
            return self._fake_state.get(key)


redis_client = StrictRedis(host="localhost", port=6379, decode_responses=True)


class RedisModel:
    def __init__(self, redis_client: StrictRedis, key: str):
        self.redis_client = redis_client
        self.key = key

    @property
    def state(self):
        return self.redis_client.get(self.key)

    @state.setter
    def state(self, value):
        self.redis_client.set(self.key, value)


class TrafficLightMachine(StateMachine):
    "A traffic light machine"

    green = State(initial=True)
    yellow = State()
    red = State()

    cycle = green.to(yellow) | yellow.to(red) | red.to(green)

Executing

redis_model = RedisModel(redis_client=redis_client, key="traffic_light")

sm = TrafficLightMachine(model=redis_model)
sm.send("cycle")

assert sm.current_state.id == "yellow"

del sm

sm = TrafficLightMachine(model=redis_model)

assert sm.current_state.id == "yellow"
assert redis_model.state == "yellow"

[shun-liang]

Many thanks!