Skip to content

Instantly share code, notes, and snippets.

@kennethreitz
Created July 6, 2012 08:41
Show Gist options
  • Save kennethreitz/3059004 to your computer and use it in GitHub Desktop.
Save kennethreitz/3059004 to your computer and use it in GitHub Desktop.
PostgreSQL hstore + SQLAlchemy
import collections
import sqlalchemy.types
class Hstore(sqlalchemy.types.UserDefinedType, sqlalchemy.types.MutableType):
"""The ``hstore`` type that stores a dictionary. It can take an any
instance of :class:`collections.Mapping`.
It can be extended to store other types than string e.g.::
class IntegerBooleanHstore(Hstore):
'''The ``hstore`` type for integer keys and boolean values.'''
def map_bind_key(self, key):
if key is not None:
return unicode(key)
def map_bind_value(self, value):
if value is not None:
return u't' if value else u'f'
def map_result_key(self, key):
if key is not None:
return int(key)
def map_result_value(self, value):
if value is not None:
return value == u't'
:param value_nullable: to prevent ``None`` (``NULL``) for dictionary
values, set it ``True``. default is ``False``
:type value_nullable: :class:`bool`
"""
def __init__(self, value_nullable=True):
self.value_nullable = bool(value_nullable)
def map_bind_key(self, key):
"""The mapping function that is used for binding keys. The default
implementation is just a string identity function.
:param key: a key object to bind
:returns: a mapped key string
:rtype: :class:`unicode`
"""
if key is None:
return
if not isinstance(key, basestring):
raise TypeError('hstore key must be a string, not ' + repr(key))
return unicode(key)
def map_bind_value(self, value):
"""The mapping function that is used for binding values.
The default implementation is just a string identity function.
:param value: a value to bind
:returns: a mapped value string
:rtype: :class:`unicode`
"""
if value is None:
return
if not isinstance(value, basestring):
raise TypeError('hstore value must be a string, not ' +
repr(value))
return unicode(value)
def map_result_key(self, key):
"""The mapping function that is used for resulting keys. The default
implementation is just an identity function.
:param key: a raw key of the result
:type key: :class:`unicode`
:returns: a mapped key object
"""
return key
def map_result_value(self, value):
"""The mapping function that is used for resulting values.
The default implementation is just an identity function.
:param key: a raw value of the result
:type key: :class:`unicode`
:returns: a mapped value
"""
return value
def get_col_spec(self):
return 'hstore'
def is_mutable(self):
return True
def compare_values(self, x, y):
x = None if x is None else dict(x)
y = None if y is None else dict(y)
return x == y
def copy_value(self, value):
if value is not None:
return dict(value)
def bind_processor(self, dialect):
def process(value):
if value is None:
return
if not isinstance(value, collections.Mapping):
raise TypeError('expected a collections.Mapping object, not '
+ repr(value))
items = getattr(value, 'iteritems', value.items)()
map_bind_key = self.map_bind_key
def map_key(key):
if key is None:
raise TypeError('hstore key cannot be None')
return map_bind_key(key)
if self.value_nullable:
map_value = self.map_bind_value
else:
map_bind_value = self.map_bind_value
def map_value(value):
if value is None:
raise TypeError('hstore value cannot be None')
return map_bind_value(value)
return dict((map_key(k), map_value(v)) for k, v in items)
return process
def result_processor(self, dialect, coltype):
def process(value):
map_key = self.map_result_key
map_value = self.map_result_value
return dict((map_key(k), map_value(v))
for k, v in value.iteritems())
return process
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment