December 31, 2020By Jason Cronquist← Back to Blog

Experimental State Management in Python

I needed a way to manage state that is loaded from a set of configs, as well as state that is dependent on those states. I thought the solution that I came up with was worth sharing.

Some useful properties of the object below:

Indicates when it needs to be rebuilt via .valid
Raises an error if the value is requested but the state has been invalidated
Recurses any dependent StateObjects used to create the state
States are only built when needed
States are defined by how they are built

    """ State Object that can be invalidated """
    from typing import (
        Any,
        Union,
        List,
        TypeVar
    )
    import threading
    from functools import reduce
    
    # pylint: disable=invalid-name
    T = TypeVar('T')
    
    class InvalidStateObjectRequested(Exception):
        """ Thrown when invalidated StateObject value is requested """
    
    class Validatable:
        """ class with valid property """
        @property
        def valid(self):
            """ Intended as Abstract method, must override """
            return False
    
    class StateObject(Validatable):
        """ manages a state object that can be invalidated """
        __recalc_required = True
        __value: Any
        __lock = threading.Lock()
        __depends_on: List[Validatable]
    
        def __init__(self, depends_on: Union[List[Validatable], Validatable, None] = None):
            if depends_on is None:
                self.__depends_on = []
            elif isinstance(depends_on, list):
                self.__depends_on = depends_on
            else:
                self.__depends_on = [depends_on]
    
        def invalidate(self):
            """ Invalidates current value.  Must be set before data can be returned """
            self.__recalc_required = True
    
        @property
        def valid(self) -> bool:
            """ Property, true if a the value is still valid.  Otherwise false """
            return not self.__recalc_required and reduce(lambda a, b: a and b, map(lambda a: a.valid, self.__depends_on), True)
    
        def set(self, value: T) -> None:
            """ Set the value, resets value to valid """
            self.__lock.acquire()
            self.__recalc_required = False
            self.__value = value
            self.__lock.release()
    
        def get(self) -> T:
            """ Returns the value, throws InvalidStateObjectRequested if invalid """
            self.__lock.acquire()
            if self.__recalc_required:
                self.__lock.release()
                raise InvalidStateObjectRequested()
            result = self.__value
            self.__lock.release()
            return result

The state then can be used like so...

from .load_state import (
    rebuild_my_state,
    rebuild_my_sub_state
)

class MyStatefulObject:
    my_state_cache: StateObject
    my_sub_state_cache: StateObject

    def __init__(self):
        self.my_state_cache = StateObject(depends_on=None)
        self.my_sub_state_cache = StateObject(depends_on=self.my_state_cache)

    @property
    def my_state(self):
        if not self.my_state_cache.valid:
            self.my_state_cache.set(rebuild_my_state())
        return self.my_state_cache.get()
    
    @property
    def my_sub_state(self):
        if not self.my_sub_state_cache.valid:
            self.my_sub_state_cache.set(rebuild_my_sub_state(self.my_state))
        return self.my_sub_state_cache.get()

    def rebuild(self, target_state: str = 'all'):
        if target_state == 'my_state' or target_state == 'all':
            self.my_state_cache.invalidate()

        if target_state == 'my_sub_state' or target_state == 'all':
            self.my_sub_state_cache.invalidate()