Python Generator DSLs
Python generators have a curious ability to both to yield values from generators (suspending computation), and receive values from a yield, which is far less known. This is equivalent to Kotlin’s coroutines and allows using them for DSLs.
If you are reading this, you presumably know about generator syntax in Python:
What is less known is how to “drive” a generator manually:
That generators can return values:
And that generators can actually accept values as “results” of yields:
def go(): x = yield 1 y = yield x z = yield y return z # Has NO equivalent in terms of loops. generator = go() first_value = False yield_result = None while True: try: if first_value: yielded = generator.__next__() else: yielded = generator.send(yield_result) print(yielded) # Prints 1, 2, 3 on separate lines # Return yielded + 1 to the generator. yield_result = yielded + 1 except StopIteration as ex: print('Result:', ex.value) # Prints "Result: 4" break
In fact, we can simplify our loop a bit since the initial
__next__() call can be replaced with
# Has NO equivalent in terms of loops. generator = go() yield_result = None while True: try: yielded = generator.send(yield_result) print(yielded) # Prints 1, 2, 3 on separate lines # Return yielded + 1 to the generator. yield_result = yielded + 1 except StopIteration as ex: print('Result:', ex.value) # Prints "Result: 4" break
Using a combination of an ad-hoc algebraic data type and generator constructs, we can make a simple monadic DSL:
Now we need some way to run it:
def run_agent(program): time = 0 yield_result = None while True: try: yielded = generator.send(yield_result) if isinstance(yielded, Log): print(yielded.message) yield_result = None if isinstance(yielded, GetTime): yield_result = time time += 1 except StopIteration as ex: return ex.value run_agent(agent()) # Prints ''' 0 -> 0 1 -> 2 2 -> 4 3 -> 6 4 -> 8 5 -> 10 6 -> 12 7 -> 14 8 -> 16 9 -> 18 '''
We can separate execution of individual action from running an entire computation:
class State(object): def __init__(self): self.time = 0 # Runs a single action against the current state. # Returns a new state and the result of the action. def run_one(state: State, action): time = state.time state.time += 1 if isinstance(action, Log): print(action.message) return state, None elif isinstance(action, GetTime): return state, time # Runs a generator against the given "action evaluation function". def run_many(state: State, generator, evaluator): yield_result = None while True: try: yielded = generator.send(yield_result) state, yield_result = evaluator(state, yielded) except StopIteration as ex: return ex.value run_many(State(), go(), run_one)
Inversion of inverted control
Often you encounter APIs that have so-called “inversion of control” - you are not controlling the main loop of the application, some other piece of code is, and you are supposed to handle “incoming events” using a fixed set of callbacks:
This is a fairly common but annoying design. Some of the issues with this design:
- All communication between callbacks has to be done through a mutable state defined on the
- Transitions between fundamentally different logical states have to be handled through some sort of hierarchy of state machines and handlers delegating messages down that hierarchy.
- Logic is spread out across multiple different callbacks. The higher the granularity of the callbacks, the more spread-out it is.
Can we “re-invert” control flow in such a situation?
Indeed we can, with the use of monadic generator DSLs:
# The only action necessary for our example. Wait = namedtuple('Wait', '') class Wrapper(Handler): def __init__(self, generator): self.generator = generator # Make generator progress to the first yield. generator.send(None) def on_event1(self, event1): generator.send(event1) def on_event2(self, event2): generator.send(event2) def on_event3(self, event3): generator.send(event3) def go(): # Imperative style loop without inversion of control. while True: event = yield Wait() if isinstance(event, Event1): ... elif isinstance(event, Event2): ... elif ...: ... run_main_loop(Wrapper(go()))