Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Redis database object test, making a forum post with replies.
import redis
import json
from juggernaut import Juggernaut
r = redis.StrictRedis()
jug = Juggernaut()
def publish_reply(fn):
def wrapper(*args, **kwargs):
channel = 'channel_' + args[0].id
fn(*args, **kwargs)
jug.publish(channel, json.dumps({'title': kwargs.get('title'), 'message': kwargs.get('message')}))
return wrapper
def check_validity(fn):
def wrapper(*args, **kwargs):
if kwargs.has_key('id'):
id = 'post:' + str(kwargs['id'])
id = args[0].id
if not r.exists(id):
raise Exception("Invalid Post ID: %s" % (id.split(':')[1]))
return fn(*args, **kwargs)
return wrapper
class Post(object):
class DataObj(object):
def __init__(self, title=None, message=None):
self.title = title
self.message = message
def __init__(self, title=None, message=None, incr='posts', id=None):
if not id:
self.incr = incr = 'post:' + str(r.incr(incr))
r.lpush(, json.dumps({'title': title, 'message': message}))
else: = 'post:' + str(id)
def get(self, id=None):
return Post(id=id)
def reply(self, title, message=None):
r.rpush(, json.dumps({'title': title, 'message': message}))
def delete(self):
def expire(self, s=5):
r.expire(, s)
def post(self):
return self.DataObj(**json.loads(r.lrange(, 0, 0)[0]))
def replies(self):
replies = r.lrange(, 1, -1)
return [self.DataObj(**json.loads(reply)) for reply in replies]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.