Skip to content

Instantly share code, notes, and snippets.

@deldotdr
Created June 10, 2010 17:51
Show Gist options
  • Save deldotdr/433353 to your computer and use it in GitHub Desktop.
Save deldotdr/433353 to your computer and use it in GitHub Desktop.
"""
@file dictstore.py
@brief Demonstration of technique for testing a class in the magnet/lcaarch
shell that requires asynchronous setup.
"""
from twisted.internet import defer
class DictStore(object):
def __init__(self, backend):
"""
@param backend an instance providing the ion.data.store.IStore
interface.
"""
self.backend = backend
@defer.inlineCallbacks
def get(self, key):
"""
@param key of stored dict
"""
serialiezed_d = yield self.backend.get(key)
d = eval(serialiezed_d)
defer.returnValue(d)
@defer.inlineCallbacks
def put(self, key, d):
"""
@brief Simple basic example for storing a dictionary
@param key key to store dict at
@param d instance of a dict
"""
assert isinstance(d, dict)
serialiezed_d = str(d)
yield self.backend.put(key, serialiezed_d)
@defer.inlineCallbacks
def test(shell_locals):
"""
@brief In the magnet shell, import test from dictstore.
Then, call dictstore.test(locals())
locals() is the built-in function that returns a dictionary of the
shell's namespace.
test returns a Deferred that fires with None, but when it does fire,
the shell name space will be *magically* updated with the objects in
the body of test:
- store
- s
- dict_store
You can now play with the instance of DictStore, called dict_store!
"""
from ion.data import store
s = yield store.Store.create_store()
dict_store = DictStore(s)
shell_locals.update(locals())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment