Though separation of concerns may be the most important design principle in software, its effective implementation is often elusive. A common problem in web design is how to link a sequence of pages together without scattering their logic all over the application. While this problem has been almost completely solved by continuation based web servers, not every language supports continuations. There is a middle ground however: coroutines. This post describes a light-weight approach to doing continuation-style web programming using Python’s coroutines.

Our target application will be the following “guess a number” game.

(simple.py) download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

#!/usr/bin/env python

import random

rgen = random.Random()

def start_game():
    name = raw_input("Greetings. What is your name? ")
    print "Hello, %s." % (name,)

    play_game()

def play_game():
    print "I am thinking of a number between 1 and 100."
    my_number = rgen.randint(1,100)
    num_guesses = 0
    while True:
        user_number = int(raw_input("Guess: "))
        num_guesses += 1
        if my_number == user_number:
            break
        elif my_number < user_number:
            print "Try lower."
        else:
            print "Try higher."

    print "Correct in %s guesses." % num_guesses
    play_again = raw_input("Play again? ")

    if play_again.startswith('y') or play_again.startswith('Y'):
        play_game()
    else:
        print "Thank you for playing."

if __name__=="__main__":
    start_game()

Here is what the program looks like using coroutines:

(game.py) download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

import random

rgen = random.Random()

class RandomNumberGame(object):
    def __init__(self, interface):
        self.interface = interface

    def __iter__(self):
        name = yield lambda: self.interface.prompt_string(
                "Greetings. What is your name? ")
        yield lambda: self.interface.display("Hello, %s." % (name,))

        # The coroutine equivalent of self.__play_game()
        iter = self.__play_game()
        try:
            y = yield iter.next()
            while True:
                y = yield iter.send(y)
        except StopIteration:
            pass

    def __play_game(self):
        yield lambda: self.interface.display("I am thinking of a number between 1 and 100.")
        my_number = yield lambda: rgen.randint(1,100)
        num_guesses = 0
        while True:
            user_number = yield lambda: self.interface.prompt_int("Guess: ")
            num_guesses += 1
            if my_number == user_number:
                break
            elif my_number < user_number:
                yield lambda: self.interface.display("Try lower.")
            else:
                yield lambda: self.interface.display("Try higher.")

        yield lambda: self.interface.display("Correct in %s guesses." % num_guesses)
        play_again = yield lambda: self.interface.prompt_yes_no("Play again? ")

        if play_again:
            # The coroutine equivalent of self.__play_game()
            iter = self.__play_game()
            try:
                y = yield iter.next()
                while True:
                    y = yield iter.send(y)
            except StopIteration:
                pass
        else:
            yield lambda: self.interface.display("Thank you for playing.")

Essentially, all read and write actions with the outside world have been replaced with the yield lambda pattern. That includes the call to rgen.randint, because rgen has been initialized according to the current time.

All we need now is an interface that implements the following methods:

# displays the given text to the user and returns None
interface.display(text)

# displays the given text to the user and returns a string input by the user
interface.prompt_string(text)

# display the given text to the user and returns an int input by the user
interface.prompt_int(text)

# display the given text to the user and returns True for yes and False for no
interface.prompt_yes_no(text)

We’ll start with the simpler command line version:

(cli.py) download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

#!/usr/bin/env python

class CommandLineInterface(object):
    def __init__(self, routine):
        self.routine = routine(self)

    def run(self):
        iter = self.routine.__iter__()
        try:
            action = iter.next()
            while True:
                action = iter.send(action())
        except StopIteration:
            pass

    def display(self, prompt):
        print prompt

    def prompt_string(self, prompt):
        return raw_input(prompt)

    def prompt_int(self, prompt):
        return int(raw_input(prompt))

    def prompt_yes_no(self, prompt):
        response = raw_input(prompt)
        return response.startswith('y') or response.startswith('Y')

if __name__=="__main__":
    from game import RandomNumberGame
    cmd = CommandLineInterface(RandomNumberGame)
    cmd.run()

The behavior of cli.py + game.py is completely identical to simple.py. Remarkably, though, the core logic of the game (in game.py) is now re-usable with any user interface supporting the four methods given above.

A typical web-MVC-style solution to the “guess a number” game would probably have a controller which dispatched on one of three different situations: the user has input her name, the user has input a guess, or the user has told us whether or not she would like to keep playing. The three different situations would likely be represented as distinct URIs. In our game.py, however, a situation corresponds to the “yield lambda” at which execution has been paused.

The essential idea to writing a coroutine-based web interface is this: only run the game routine up to the point where more information is needed. Store the result of every lambda yielded so far. On successive page requests, replay the routine with the stored results, but only invoke the lambdas that were not invoked on a previous page request. The medium for storing the results of the lambdas does not matter. It could be embedded in hidden input elements in HTML (though this raises issues of trust), or stored in a database tied to a session ID. For simplicity, the following implementation stores the values in memory, tied to a value stored in a hidden input element.

(web.py) download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

#!/usr/bin/env python

import cgi
from wsgiref.simple_server import make_server

# Everytime we generate an html page, we create a hidden input element
# with this name. It lets us know which saved history must be resumed
# in order to continue the program.
HISTORY_FORM_NAME = 'WebInterface_history_index'

class History(object):
    """A record of past values returned by the routine. Useful for playback.
       self.get_pending is a function which takes a dictionary like object
           corresponding to the POST variables, and returns the new value
           to be added to the history.
       self.old_values is all of the results of functions yielded by the routine.
       """
    def __init__(self, get_pending, old_values):
        self.get_pending = get_pending
        self.old_values = old_values

class WebInterface(object):
    """WebInterface wraps around a routine class to allow for the routine to
       be executed through a browser. It works by remembering the results
       of functions that were yielded by the routine. In order to pick up
       where it left off, the routine is re-run with the remembered values,
       a new value is parsed from the POST variables, and the routine keeps
       running until it reaches another prompt."""
    def __init__(self, routine_class):
        self.routine_class = routine_class
        self.histories = [History(None, [])]

    def respond(self, environ, start_response):
        responder = WebInterface.Response(self, environ, start_response)
        return responder.respond()

    class Response(object):
        """The Response class is instantiated for every HTTP request.
           It grabs the appropriate history according to the value in the POST
           variable 'WebInterface_history_index'. Using that history, it
           re-runs the routine with the contained old_values, parses the POST
           variables for any new values using get_pending, and continues the
           routine until it needs to make another request. For simplicity,
           it modifies web_interface.histories directly. A real
           implementation would need to protect this variable using locks
           (or find a mutation-free solution), since it is possible for
           the wsgi server to make concurrent calls to WebInterface.respond.
           """
        def __init__(self, web_interface, environ, start_response):
            self.web_interface = web_interface
            self.environ = environ
            self.start_response = start_response

        def respond(self):
            routine = self.web_interface.routine_class(self)
            self.form = cgi.FieldStorage(fp=self.environ['wsgi.input'],environ=self.environ)
            history_index = int(self.form.getvalue(HISTORY_FORM_NAME, default="0"))
            if len(self.web_interface.histories) <= history_index:
                self.start_response('412 Cannot read the future', [('Content-type', 'text/html')])
                return ["That history has not yet been written."]

            # Copy the history in order to create a new history.
            history_orig = self.web_interface.histories[history_index]
            self.history = History(history_orig.get_pending, history_orig.old_values[:])
            self.start_response('200 OK', [('Content-type', 'text/html')])
            self.output = [ '<form method="POST">\n' ]

            self.paused = False
            iter = routine.__iter__()
            try:
                # Re-run the routine over all old_values.
                action = iter.next()
                for history_value in self.history.old_values:
                    action = iter.send(history_value)

                # If a get_pending was previously set, invoke it in order
                # to parse the POST variables for any new values.
                if self.history.get_pending != None:
                    val = self.history.get_pending(self)
                    self.history.old_values.append(val)
                    action = iter.send(val)

                # Continue the routine until another prompt is made.
                while not self.paused:
                    new_value = action()
                    if not self.paused:
                        self.history.old_values.append(new_value)
                        action = iter.send(new_value)

            except StopIteration:
                pass

            self.web_interface.histories.append(self.history)
            self.output += '<input type="hidden" name="%s" value="%s">\n' % (HISTORY_FORM_NAME, len(self.web_interface.histories) - 1)
            self.output += '</form>\n'
            return self.output

        def display(self, str):
            self.output += str + "
\n"

        def prompt_string(self, prompt):
            self.prompt_type(prompt, str)

        def prompt_int(self, prompt):
            self.prompt_type(prompt, int)

        def prompt_yes_no(self, prompt):
            self.prompt_and_pause(
                    prompt,
                    [("submit", "btn_yes", "Yes"), ("submit", "btn_no", "No")],
                    lambda form: form.has_key('btn_yes'))

        def prompt_type(self, prompt, type_parse):
            self.prompt_and_pause(
                    prompt,
                    [("text", "prompt", ""), ("submit", "btn_submit", "Enter")],
                    lambda form: type_parse(form["prompt"].value))

        def prompt_and_pause(self, prompt, inputs, parse_form):
            self.output += prompt
            for input in inputs:
                self.output += '<input type="%s" name="%s" value="%s">\n' % input
            self.paused = True
            def read_from_form(responder):
                return parse_form(responder.form)
            self.history.get_pending = read_from_form

if __name__=="__main__":
    from game import RandomNumberGame
    interface = WebInterface(RandomNumberGame)
    httpd = make_server('', 8000, interface.respond)
    httpd.serve_forever()