state_chain.py

The state_chain module helps define and run algorithms composed of multiple functions that operate on a shared state object.

Installation

state_chain is available on GitHub and on PyPI:

$ pip install state_chain

The version of state_chain documented here has been tested against Python 3.6, 3.7, 3.8 and 3.9 on Ubuntu.

state_chain is MIT-licensed.

Tutorial

This module provides an abstraction for implementing arbitrary algorithms as a list of functions that operate on a shared state object. Algorithms defined this way are easy to arbitrarily modify at run time, and they provide cascading exception handling.

To get started, create a StateChain object:

>>> from state_chain import StateChain
>>> chain = StateChain()

And add some functions to it:

>>> @chain.add
... def set_x(state):
...     state.x = 1
...
>>> @chain.add
... def set_y(state):
...     state.y = 2
...
>>> @chain.add
... def set_sum(state):
...     state.sum = state.x + state.y
...

As you can see, each function will receive the state object as its only argument. Moreover you may have noticed that the functions don’t return anything. Returning a value isn’t prohibited, but that value will be ignored by the run method.

Speaking of the run method, let’s give it a go:

>>> chain.run().sum
3

Okay, we have the expected sum!

Modifying a State Chain

Let’s define three more functions to add to the state chain:

>>> def uh_oh(state):
...     if state.x == 0:
...         raise Exception('oops, state.x is zero')
...
>>> def deal_with_it(state):
...     print("I am dealing with it!")
...     state.exception = None
...
>>> def print_state(state):
...     print(state)
...

and make a copy of the chain that we’ll use later:

>>> chain_copy = chain.copy()

Now let’s interpolate the new functions into our state chain. Let’s put the uh_oh at the beginning:

>>> chain.add(uh_oh, position=0)
<function uh_oh ...>
>>> chain.functions
(<function uh_oh ...>, <function set_x ...>, <function set_y ...>,
 <function set_sum ...>)

Then let’s remove set_y and replace set_sum with print_state:

>>> chain.remove('set_y')
>>> chain.add(print_state, position=chain.before('set_sum'), exception='accepted')
<function print_state ...>
>>> chain.remove('set_sum')
>>> chain.functions
(<function uh_oh ...>, <function set_x ...>, <function print_state ...>)

Finally, let’s add our exception handler after print_state:

>>> chain.add(deal_with_it, position=chain.after('print_state'), exception='required')
<function deal_with_it ...>
>>> chain.functions
(<function uh_oh ...>, <function set_x ...>, <function print_state ...>,
 <function deal_with_it ...>)

Note: when making extensive changes to a state chain, you can use the modify method to rebuild the entire chain in a safe way. We could have achieved the same result as above like so:

>>> chain = (
...     chain_copy.modify()
...     .add(uh_oh)
...     .keep('set_x')
...     .drop('set_y')
...     .replace('set_sum', print_state, exception='accepted')
...     .add(deal_with_it, exception='required')
...     .end()
... )
>>> chain.functions
(<function uh_oh ...>, <function set_x ...>, <function print_state ...>,
 <function deal_with_it ...>)

This allows you to see exactly what your chain does and how it differs from the original chain.

Either way, what happens when we run it?

>>> from state_chain import Object
>>> state = chain.run(Object(x=0))
Object(x=0, exception=Exception('oops, state.x is zero'))
I am dealing with it!

Exception Handling

Whenever a function raises an exception, like uh_oh did in the example above, run captures the exception and assigns it to state.exception. As long as this state attribute is not None, any normal function is skipped, and only exception handling functions get called. It’s like a fast-forward. So in our example print_state and deal_with_it got called, but set_x didn’t.

If we run without triggering the exception in uh_oh, then we have a different result:

>>> _ = chain.run(Object(x=5))
Object(x=1, exception=None)

If we remove the deal_with_it function, then the exception isn’t handled, so it’s reraised at the end of the chain:

>>> chain.remove('deal_with_it')
>>> chain.run(Object(x=0))
Traceback (most recent call last):
    ...
Exception: oops, state.x is zero

Whether a function is skipped or called is determined by its “exception preference” (the value of the exception argument of the StateChain.add method). If it’s ‘unwanted’, then the function will be skipped when an exception has been raised. If it’s ‘accepted’, then the function will always be called. If it’s ‘required’, then the function will only be called when an exception has been raised.

The default value is ‘unwanted’, but you can change it when creating the chain:

>>> chain = StateChain(exception_preference='accepted')

In that case, the chain’s functions are always called, unless they were explicitly added with a different exception preference:

>>> chain.add(uh_oh)
<function uh_oh at ...>
>>> @chain.add(exception='unwanted')
... def skipped(state):
...     raise Exception("this function should not be called")
...
>>> @chain.add
... def always_called(state):
...     state.x = -1
...     state.exception = None
...
>>> chain.run().x
-1

Argument Injection

So far we’ve only used chain functions that expect the state object as their only argument, but the run method can also automatically pass the value of any attribute of the state object to a function as an argument.

For example:

>>> def print_sum(x, y):
...     print(f"x + y = {x + y}")
...
>>> chain = StateChain(functions=[set_x, set_y, print_sum])
>>> _ = chain.run()
x + y = 3

If a chain function has an exception argument, then its default exception preference is ‘accepted’ if the argument is optional (e.g. exception=None), and ‘required’ otherwise.

>>> def exception_handler(foo, exception):
...     "This function's default exception preference is 'required'"
...
>>> def exception_tolerant_function(foo, exception=None):
...     "This function's default exception preference is 'accepted'"
...

Argument injection is implemented in the call function and relies on the standard library function inspect.signature introduced in Python 3.3.

Static Typing

Since version 2.0, the state_chain module includes complete type annotations, and its API has been redesigned to facilitate type checking the applications that use it.

Here is an example of a statically typed chain:

>>> from dataclasses import dataclass
>>> from typing import Optional
>>> @dataclass
... class State:
...     request: str
...     response: Optional[str] = None
...     exception: Optional[Exception] = None
...
>>> def respond(state: State):
...     if state.request.startswith("I want chocolate"):
...         state.response = "Me too."
...     else:
...         state.response = "Sorry, I don't understand your request."
...
>>> def match_punctuation(state: State):
...     if state.response and state.request.endswith('!'):
...         state.response = state.response.replace('.', '!')
...
>>> chain = StateChain(State, [respond, match_punctuation])
>>> chain.run(State("I want chocolate!")).response
'Me too!'

Aliases

If you’re building a state chain that is meant to be used and modified by other programs, it can be useful to give friendly and stable aliases to some of your chain’s functions.

>>> temp_chain = StateChain(State)
>>> temp_chain.add(set_y, alias='y_is_available')
<function set_y ...>

This makes it easy to insert a function before or after a specific function is run, even if that function is later renamed.

>>> temp_chain.add(print_state, position=temp_chain.after('y_is_available'))
<function print_state ...>

API Reference

Migrating from 1.x

Version 2.0 of the state_chain module includes several breaking changes.

1) The ways to create and initialize a state chain have changed. The StateChain.from_dotted_name constructor no longer exists, and the default StateChain constructor no longer takes a variable number of arguments.

-chain = StateChain.from_dotted_name(...)
+chain = StateChain()
+
+@chain.add
+def foo(...):
+    ...
+
+@chain.add
+def bar(...):
+    ...
-chain = StateChain(foo, bar)
+chain = StateChain(functions=[foo, bar])

2) The StateChain.run method no longer accepts a variable number of arguments.

-chain.run(x=0, _raise_immediately=True, _return_after='foo')
+from state_chain import Object
+chain.run(Object(x=0), raise_immediately=True, return_after='foo')

3) Modifying the state by returning dictionaries is no longer supported. You have to explicitly modify the state object instead.

-def foo():
-    return {'x': 0}
+def foo(state):
+    state.x = 0