Skip to content

Instantly share code, notes, and snippets.

Embed
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'])
else:
id = args[0].id
if not r.exists(id):
raise Exception("Invalid Post ID: %s" % (id.split(':')[1]))
else:
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
self.id = 'post:' + str(r.incr(incr))
r.lpush(self.id, json.dumps({'title': title, 'message': message}))
else:
self.id = 'post:' + str(id)
@classmethod
@check_validity
def get(self, id=None):
return Post(id=id)
@publish_reply
@check_validity
def reply(self, title, message=None):
r.rpush(self.id, json.dumps({'title': title, 'message': message}))
@check_validity
def delete(self):
r.delete(self.id)
@check_validity
def expire(self, s=5):
r.expire(self.id, s)
@property
@check_validity
def post(self):
return self.DataObj(**json.loads(r.lrange(self.id, 0, 0)[0]))
@property
@check_validity
def replies(self):
replies = r.lrange(self.id, 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.