Create a gist now

Instantly share code, notes, and snippets.

Stateful Model (2 files): a subclass of Google App Engine db.Model class that functions as a state machine. unit test for StatefulModel.
# -*- coding: utf-8 -*-
Google App Engine Stateful Model
Subclass App Engine db.Model class so that it may function as a flexible
state machine.
import logging
from google.appengine.ext import db
class StateError(Exception): pass
class StatefulModel(db.Model):
# default model properties
state = db.StringProperty(multiline=False)
created = db.DateTimeProperty(auto_now_add=True)
updated = db.DateTimeProperty(auto_now=True)
# define states and actions
states = {
'sample' : {
'enter': lambda record: logging.debug('entering sample state'),
'exit': lambda record: logging.debug('exiting sample state'),
'persist': lambda record: record
def __call__(self, action):
"""calls current state action"""
if action in self.states[self.state]:
return self.states[self.state][action](self)
raise StateError("invalid action '%s' request for state: %s", (
action, self.state))
def change_state(self, new_state, put_to_datastore=True):
"""record is saved by default whenever state changed"""
old_state = self.state
if old_state in self.states and 'exit' in self.states[old_state]:
if new_state in self.states and 'enter' in self.states[new_state]:
elif new_state in self.states:
raise StateError('state %s not set in states dict' % (new_state))
self.state = new_state
logging.debug('change state: %s to %s' % (old_state, new_state))
if put_to_datastore:
return self
def new(ModelClass, state):
"""Factory method to instantiate new class. As a rule, this should
be overridden to accommodate additional arguments."""
record = ModelClass()
record.change_state(state, False)
return record
def find_by_state(ModelClass, state):
"""returns an iterable"""
gql = "WHERE state = :state"
records = ModelClass.gql(gql, state=state).run()
return records
StatefulModel Test
# Python Imports
import logging
import unittest
# App Engine Imports
from google.appengine.ext import (db)
# Project Imports
from project.models.stateful_model import (StatefulModel, StateError)
# Dev Objects
class ThreatLevel(StatefulModel):
# additional properties
output = db.StringProperty(multiline=False)
# states and actions
states = {
'green': {
'enter': lambda record: setattr(record, 'output',
"everything's cool"),
'elevate': lambda record: record.change_state('yellow'),
'lower': lambda record: record
'yellow': {
'enter': lambda record: lambda record: setattr(record, 'output',
"getting tense"),
'elevate': lambda record: record.change_state('red'),
'lower': lambda record: record.change_state('green')
'red': {
'enter': lambda record: record.panic(),
'exit' : lambda record: lambda record: setattr(record, 'output',
"downgrading from red"),
'elevate': lambda record: record,
'lower': lambda record: record.change_state('yellow')
# additional methods
def panic(self):
self.output = 'EVERYBODY PANIC!'
def elevate(self):
return self('elevate')
def lower(self):
return self('lower')
class PatientCondition(StatefulModel):
name = db.StringProperty(multiline=False)
states = {
'stable': {
'downgrade': lambda record: record.change_state('serious'),
'upgrade': lambda record: logging.error("can't upgrade from stable")
'serious': {
'downgrade': lambda record: record.change_state('critical'),
'upgrade': lambda record: record.change_state('stable'),
'critical': {
'downgrade': lambda record: logging.error(
"can't upgrade from critical"),
'upgrade': lambda record: record.change_state('serious'),
def downgrade(self):
return self('downgrade')
def upgrade(self):
return self('upgrade')
def new(ModelClass, name, condition):
"""Factory method to instantiate new class."""
record = ModelClass(name=name)
record.change_state(condition) # will be autosaved
return record
# Test Class
class StatefulModelDevTest(unittest.TestCase):
# Harness
def setUp(self):
def tearDown(self):
# Test
def testStateError(self):
patient ='Martin', 'stable')
self.assertRaises(StateError, patient.change_state, 'fantastic')
self.assertRaises(StateError, patient, 'deceased')
def testPatientConditionModel(self):
patient ='John', 'serious')
# downgrade (will be autosaved)
patient_record = list(PatientCondition.find_by_state('critical'))[0]
self.assertEqual(patient_record.state, 'critical')
self.assertEqual(, 'John')
# change state (without autosaving)
autosave = False
patient.change_state('stable', autosave)
stable_count = len(list(PatientCondition.find_by_state('stable')))
critical_count = PatientCondition.gql("WHERE state = :state",
self.assertEqual(patient.state, 'stable')
self.assertEqual(stable_count, 0)
self.assertEqual(critical_count, 1)
def testThreatLevelModel(self):
threat ='yellow')
self.assertEqual(threat.state, 'red')
saved_threat = list(ThreatLevel.find_by_state('red'))[0]
self.assertEqual(saved_threat.state, 'red')
self.assertEqual(saved_threat.output, 'EVERYBODY PANIC!')
self.assertEqual(saved_threat.state, 'yellow')
# Smoke Tests
def testInstance(self):
self.assertTrue(isinstance(self, unittest.TestCase))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment