Last active
August 29, 2015 14:04
-
-
Save alukach/b6d0c52868cf0caba36d to your computer and use it in GitHub Desktop.
Django Hstore Mixin - Support storing all standard data types within Hstore.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import datetime | |
import json | |
from django.db import models | |
from django_hstore import hstore | |
serialize = lambda d: ( | |
json.dumps(d.isoformat()) if isinstance(d, datetime.datetime) else | |
json.dumps(d) | |
) | |
class JsonDict(dict): | |
""" A dict-like object where all values are serialized to and from | |
JSON upon setting and getting, allowing it to be stored in and | |
retrieved from Hstore field while maintaining original type. """ | |
def __init__(self, data, modelInstance, *args, **kwargs): | |
data = data or {} | |
self._modelInstance = modelInstance | |
return super(self.__class__, self).__init__(data, *args, **kwargs) | |
def __setitem__(self, k, v): | |
# Serialize value being set | |
super(self.__class__, self).__setitem__(k, serialize(v)) | |
# You can't use __setitem__ on a @property (ie. "e.data['foo'] = 1" | |
# will return e._data as an object which then gets updated, but those | |
# updates won't # make their way back to the db). As such, we must | |
# update the db field hiding behind the property manually. | |
self._modelInstance._data = self.copy() | |
def __getitem__(self, key): | |
# Deserialize on 'get' | |
return json.loads(super(self.__class__, self).__getitem__(key)) | |
def __getattr__(self, key): | |
# Deserialize on 'get' | |
return json.loads(super(self.__class__, self).__getattr__(key)) | |
def __repr__(self): | |
# Deserialize values for reproduction | |
return repr({k: v for k, v in self.items()}) | |
def __lt__(self, other): | |
return {k: v for k, v in self.items()} < other | |
def __le__(self, other): | |
return {k: v for k, v in self.items()} <= other | |
def __eq__(self, other): | |
return {k: v for k, v in self.items()} == other | |
def __ne__(self, other): | |
return {k: v for k, v in self.items()} != other | |
def __gt__(self, other): | |
return {k: v for k, v in self.items()} > other | |
def __ge__(self, other): | |
return {k: v for k, v in self.items()} <= other | |
def get(self, key, default=None): | |
value = super(self.__class__, self).get(key, default) | |
# Only deserialize if value was pulled from dict | |
if value and value != default: | |
value = json.loads(value) | |
return value | |
def items(self): | |
return [(k, json.loads(v)) for k, v in super(self.__class__, self).items()] | |
class HstoreMixin(models.Model): | |
""" Data field to be added to model to enable Hstore field. Actual | |
hstore field hidden with underscore, property field serializes and | |
deserializes data upon setting/getting. """ | |
_data = hstore.DictionaryField( | |
'KeyValueStore', | |
db_index=True, null=True, blank=True | |
) | |
objects = hstore.HStoreManager() | |
class Meta: | |
abstract = True | |
@property | |
def data(self): | |
""" Decode data from JSON """ | |
return JsonDict(self._data, modelInstance=self) | |
@data.setter | |
def data(self, value): | |
""" Encode data to JSON """ | |
if not self._data: | |
self._data = {k: serialize(v) for k, v in value.items()} | |
else: | |
self._data = JsonDict(value, modelInstance=self) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import datetime | |
import json | |
from django.test import TestCase | |
from YOURAPP.models import SomeObjectUsingHstoreMixin | |
class TestHstoreMixin(TestCase): | |
def setUp(self): | |
self.now = datetime.datetime.now() | |
# TODO: Adjust this to match your model | |
self.e = SomeObjectUsingHstoreMixin.objects.create( | |
data=dict( | |
int=1, | |
string='foo', | |
date=self.now, | |
list=[1,'two'], | |
dict=dict(a=1) | |
) | |
) | |
def test_setting(self): | |
self.assertEqual( | |
self.e._data, | |
{ | |
'date': json.dumps(self.now.isoformat()), | |
'int': '1', | |
'list': '[1, "two"]', | |
'string': '"foo"', | |
'dict': '{"a": 1}' | |
} | |
) | |
self.e.data['int'] = 3 | |
self.e.save() | |
self.assertEqual(self.e._data['int'], '3') | |
self.e.data['list'] = [2, 'three'] | |
self.e.save() | |
self.assertEqual(self.e._data['list'], '[2, "three"]') | |
self.e.data['string'] = 'Foo' | |
self.e.save() | |
self.assertEqual(self.e._data['string'], '"Foo"') | |
self.e.data['dict'] = dict(a=2) | |
self.e.save() | |
self.assertEqual(self.e._data['dict'], '{"a": 2}') | |
def test_getting(self): | |
self.assertEqual( | |
self.e.data, | |
dict(int=1, string='foo', date=self.now.isoformat(), list=[1, 'two'], dict={'a': 1}) | |
) | |
self.assertEqual(self.e.data.get('int'), 1) | |
self.assertEqual(self.e.data.get('string'), 'foo') | |
self.assertEqual(self.e.data.get('date'), self.now.isoformat()) | |
self.assertEqual(self.e.data.get('list'), [1, 'two']) | |
self.assertEqual(self.e.data.get('dict'), {'a': 1}) | |
self.assertEqual(self.e.data.get('notThere'), None) | |
self.assertEqual(self.e.data.get('notThere', 'Nope'), 'Nope') | |
self.assertEqual(self.e.data['int'], 1) | |
self.assertEqual(self.e.data['string'], 'foo') | |
self.assertEqual(self.e.data['date'], self.now.isoformat()) | |
self.assertEqual(self.e.data['list'], [1, 'two']) | |
self.assertEqual(self.e.data['dict'], {'a': 1}) | |
with self.assertRaises(KeyError): | |
self.e.data['notThere'] | |
def test_comparison(self): | |
self.assertTrue( | |
self.e.data == dict( | |
int=1, | |
string='foo', | |
date=self.now.isoformat(), | |
list=[1, 'two'], | |
dict={'a': 1} | |
) | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment