Skip to content

Instantly share code, notes, and snippets.

@voidfiles
Created December 28, 2013 06:31
Show Gist options
  • Select an option

  • Save voidfiles/8156685 to your computer and use it in GitHub Desktop.

Select an option

Save voidfiles/8156685 to your computer and use it in GitHub Desktop.
I was hoping that I could just easily store some models in redis, but then ended up going with a solution that I didn't build, before I jumped ship though I ended up with a fully fledged redis model system. I just didn't want to delete so I put it up as a gist.
# The MIT License (MIT)
#
# Copyright (c) 2013 Alex Kessinger
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import json
import redis
r_con = redis.Redis.from_url('redis://localhost:6379/11')
class StuctRedisModel(dict):
key_prefix = None
valid_attrs = tuple()
indexes = tuple()
list_indexes = tuple()
custom_indexes = tuple()
def __setattr__(self, name, val):
return self.__setitem__(name, val)
def __getattr__(self, name):
try:
return self.__getitem__(name)
except KeyError:
raise AttributeError(name)
def __init__(self, r_con, **kwargs):
self.r_con = r_con
for attr in self.valid_attrs:
self[attr] = kwargs.get(attr)
def update(self, data):
updated = False
for attr in data.keys():
new = data.get(attr)
old = self.get(attr)
if old != new:
updated = True
if attr in self.indexes:
self.del_index(attr)
if attr in self.list_indexes:
self.del_list_index(attr)
for index in self.custom_indexes:
if attr in index[1]:
self.del_index(index)
self[attr] = data.get(attr)
for index in self.custom_indexes:
if attr in index[1]:
self.add_index(index[0])
if attr in self.indexes:
self.add_index(attr)
if attr in self.list_indexes:
self.add_list_index(attr)
if updated:
self.save()
@classmethod
def from_data(cls, r_con, data):
kwargs = json.loads(data)
return cls(r_con, **kwargs)
def to_dict(self):
data = {}
for attr in self.valid_attrs:
data[attr] = self.get(attr)
return data
def items(self):
return self.to_dict().items()
def iteritems(self):
return self.to_dict().iteritems()
def serialize(self):
data = self.to_dict()
return json_dumps.encode(data)
def for_json(self):
return self.to_dict()
def add_index(self, index):
self.r_con.set('%s:%s:%s' % (self.key_prefix, index, getattr(self, index)), self.pk)
def del_index(self, index):
self.r_con.delete('%s:%s:%s' % (self.key_prefix, index, self.get(index)))
def add_list_index(self, index):
self.r_con.lpush('%s:%s:%s' % (self.key_prefix, index, getattr(self, index)), unicode(self.pk))
def del_list_index(self, index):
self.r_con.lrem('%s:%s:%s' % (self.key_prefix, index, self.get(index)), unicode(self.pk), 0)
def delete(self):
if not self.pk:
return self
key_root = '%s' % (self.key_prefix)
self.r_con.delete('%s:%s' % (key_root, self.pk))
for index in self.indexes:
self.del_index(index)
for index in self.list_indexes:
self.del_list_index(index)
return self
def save(self):
creating = False
if not self.pk:
creating = True
pk = self.r_con.incr('%s:_meta:pk' % self.key_prefix)
self.pk = pk
data = self.serialize()
key_root = '%s' % (self.key_prefix)
self.r_con.set('%s:%s' % (key_root, self.pk), data)
if creating:
for index in self.indexes:
self.add_index(index)
for index in self.list_indexes:
self.add_list_index(index)
def __repr__(self):
return self.serialize()
@classmethod
def get_by_index(cls, r_con, index, *args):
if index not in cls.indexes:
raise Exception('%s not in indexes %s' % (index, cls.indexes))
args = ':'.join(map(unicode, args))
key = '%s:%s:%s' % (cls.key_prefix, index, args)
pk = r_con.get(key)
if pk:
return cls.by_pk(r_con, pk)
return None
@classmethod
def get_list_by_index(cls, r_con, index, *args):
if index not in cls.list_indexes:
raise Exception('%s not in indexes %s' % (index, cls.list_indexes))
args = ':'.join(map(unicode, args))
key = '%s:%s:%s' % (cls.key_prefix, index, args)
pks = map(int, r_con.lrange(key, 0, -1))
if pks:
items = cls.by_pks(r_con, pks)
return items
@classmethod
def by_pk(cls, r_con, pk):
key = '%s:%s' % (cls.key_prefix, pk)
data = r_con.get(key)
if data:
return cls.from_data(r_con, data)
return data
@classmethod
def by_pks(cls, r_con, pks):
keys = ['%s:%s' % (cls.key_prefix, pk) for pk in pks]
items = r_con.mget(keys)
if items:
return [cls.from_data(r_con, item) for item in items]
return items
if __name__ == '__main__':
class TestModel(StuctRedisModel):
key_prefix = 'test'
valid_attrs = (
'pk',
'attr',
'fk',
)
indexes = (
'attr',
'custom_index',
)
list_indexes = (
'fk',
)
custom_indexes = (
('custom_index', ('attr', )),
)
@property
def custom_index(self):
return '%s:b' % (self.attr)
model_data = {
'attr': 'awesome',
'fk': 1,
}
model_data_with_extra = {
'non_attr': 'not_cool'
}
model_data_with_extra.update(model_data)
test_model = TestModel(r_con, **model_data_with_extra)
assert test_model.attr == 'awesome'
assert not hasattr(test_model, 'not_cool')
model_data = {
'pk': None,
'attr': 'awesome',
'fk': 1,
}
test_model = TestModel(r_con, **model_data)
assert test_model.to_dict() == model_data
assert test_model.serialize() == '{"pk": null, "fk": 1, "attr": "awesome"}'
test_model.save()
assert r_con.get('test:1') == '{"pk": 1, "fk": 1, "attr": "awesome"}'
assert r_con.get('test:attr:awesome') == '1'
assert TestModel.by_pk(r_con, 1).attr == 'awesome'
assert TestModel.get_by_index(r_con, 'attr', 'awesome').attr == 'awesome'
assert TestModel.get_by_index(r_con, 'custom_index', 'awesome:b').attr == 'awesome'
test_model.delete()
assert r_con.get('test:1') is None
assert r_con.get('test:attr:awesome') is None
model_data_1 = {
'pk': None,
'attr': 'awesome',
'fk': 1,
}
model_data_2 = {
'pk': None,
'attr': 'awesome2',
'fk': 1,
}
test_model1 = TestModel(r_con, **model_data_1)
test_model1.save()
test_model2 = TestModel(r_con, **model_data_2)
test_model2.save()
test_models = TestModel.by_pks(r_con, [test_model1.pk, test_model2.pk])
assert len(test_models) == 2
test_models = TestModel.get_list_by_index(r_con, 'fk', 1)
assert len(test_models) == 2
model_data_2 = {
'attr': 'awesome3',
'fk': 2,
}
test_model2.update(model_data_2)
test_models = TestModel.get_list_by_index(r_con, 'fk', 1)
assert len(test_models) == 1
test_models = TestModel.get_list_by_index(r_con, 'fk', 2)
assert len(test_models) == 1
test_model = TestModel.get_by_index(r_con, 'attr', 'awesome3')
assert test_model.fk == 2
assert test_model.custom_index == 'awesome3:b'
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment