# Python Generator DSLs

Last modified on November 2, 2020 by Alex

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.

## Python Generators

If you are reading this, you presumably know about generator syntax in Python:

def go():
yield 1
yield 2
yield 3

for x in go():
print(x) # Prints 1, 2, 3 on separate lines
print([x for x in go()]) # Prints [1, 2, 3]

What is less known is how to “drive” a generator manually:

# Equivalent to
#   for x in go(): print(x)
# loop.
generator = go()
while True:
try:
yielded = generator.__next__()
print(yielded) # Prints 1, 2, 3 on separate lines
except StopIteration as ex:
break

That generators can return values:

def go():
yield 1
yield 2
yield 3
return 4

# Has NO equivalent in terms of loops.
generator = go()
while True:
try:
yielded = generator.__next__()
print(yielded) # Prints 1, 2, 3 on separate lines
except StopIteration as ex:
print('Result:', ex.value) # Prints "Result: 4"
break

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 send(None),

# 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

## Monadic DSLs

Using a combination of an ad-hoc algebraic data type and generator constructs, we can make a simple monadic DSL:

Log = namedtuple('Log', 'message')
GetTime = namedtuple('GetTime', '')

def go():
for i in range(10):
t = yield GetTime()
yield Log(f'{i} -> {t}')

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:

class Handler:
def on_event1(self, event1): pass
def on_event2(self, event2): pass
def on_event3(self, event3): pass

run_main_loop(Handler())

This is a fairly common but annoying design. Some of the issues with this design:

1. All communication between callbacks has to be done through a mutable state defined on the Handler.
2. 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.
3. 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()))