SQLAlchemy JSON as blob/text

Question

I'm storing JSON down as blob/text in a column using MySQL. Is there a simple way to convert this into a dict using python/SQLAlchemy?

Answer 1

You can very easily create your own type with SQLAlchemy

---

For SQLAlchemy versions >= 0.7, check out Yogesh's answer below

---

import jsonpickle
import sqlalchemy.types as types

class JsonType(types.MutableType, types.TypeDecorator):    
    impl = types.Unicode

    def process_bind_param(self, value, engine):
        return unicode(jsonpickle.encode(value))

    def process_result_value(self, value, engine):
        if value:
            return jsonpickle.decode(value)
        else:
            # default can also be a list
            return {}

This can be used when you are defining your tables (example uses elixir):

from elixir import *
class MyTable(Entity):
    using_options(tablename='my_table')
    foo = Field(String, primary_key=True)
    content = Field(JsonType())
    active = Field(Boolean, default=True)

You can also use a different json serialiser to jsonpickle.

Answer 2

sqlalchemy.types.MutableType has been deprecated (v0.7 onward), the documentation recommends using sqlalchemy.ext.mutable instead.

I found a Git gist by dbarnett that I have tested for my usage. It has worked well so far, for both dictionary and lists.

Pasting below for posterity:

import simplejson
import sqlalchemy
from sqlalchemy import String
from sqlalchemy.ext.mutable import Mutable

class JSONEncodedObj(sqlalchemy.types.TypeDecorator):
    """Represents an immutable structure as a json-encoded string."""

    impl = String

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = simplejson.dumps(value)
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = simplejson.loads(value)
        return value

class MutationObj(Mutable):
    @classmethod
    def coerce(cls, key, value):
        if isinstance(value, dict) and not isinstance(value, MutationDict):
            return MutationDict.coerce(key, value)
        if isinstance(value, list) and not isinstance(value, MutationList):
            return MutationList.coerce(key, value)
        return value

    @classmethod
    def _listen_on_attribute(cls, attribute, coerce, parent_cls):
        key = attribute.key
        if parent_cls is not attribute.class_:
            return

        # rely on "propagate" here
        parent_cls = attribute.class_

        def load(state, *args):
            val = state.dict.get(key, None)
            if coerce:
                val = cls.coerce(key, val)
                state.dict[key] = val
            if isinstance(val, cls):
                val._parents[state.obj()] = key

        def set(target, value, oldvalue, initiator):
            if not isinstance(value, cls):
                value = cls.coerce(key, value)
            if isinstance(value, cls):
                value._parents[target.obj()] = key
            if isinstance(oldvalue, cls):
                oldvalue._parents.pop(target.obj(), None)
            return value

        def pickle(state, state_dict):
            val = state.dict.get(key, None)
            if isinstance(val, cls):
                if 'ext.mutable.values' not in state_dict:
                    state_dict['ext.mutable.values'] = []
                state_dict['ext.mutable.values'].append(val)

        def unpickle(state, state_dict):
            if 'ext.mutable.values' in state_dict:
                for val in state_dict['ext.mutable.values']:
                    val._parents[state.obj()] = key

        sqlalchemy.event.listen(parent_cls, 'load', load, raw=True, propagate=True)
        sqlalchemy.event.listen(parent_cls, 'refresh', load, raw=True, propagate=True)
        sqlalchemy.event.listen(attribute, 'set', set, raw=True, retval=True, propagate=True)
        sqlalchemy.event.listen(parent_cls, 'pickle', pickle, raw=True, propagate=True)
        sqlalchemy.event.listen(parent_cls, 'unpickle', unpickle, raw=True, propagate=True)

class MutationDict(MutationObj, dict):
    @classmethod
    def coerce(cls, key, value):
        """Convert plain dictionary to MutationDict"""
        self = MutationDict((k,MutationObj.coerce(key,v)) for (k,v) in value.items())
        self._key = key
        return self

    def __setitem__(self, key, value):
        dict.__setitem__(self, key, MutationObj.coerce(self._key, value))
        self.changed()

    def __delitem__(self, key):
        dict.__delitem__(self, key)
        self.changed()

class MutationList(MutationObj, list):
    @classmethod
    def coerce(cls, key, value):
        """Convert plain list to MutationList"""
        self = MutationList((MutationObj.coerce(key, v) for v in value))
        self._key = key
        return self

    def __setitem__(self, idx, value):
        list.__setitem__(self, idx, MutationObj.coerce(self._key, value))
        self.changed()

    def __setslice__(self, start, stop, values):
        list.__setslice__(self, start, stop, (MutationObj.coerce(self._key, v) for v in values))
        self.changed()

    def __delitem__(self, idx):
        list.__delitem__(self, idx)
        self.changed()

    def __delslice__(self, start, stop):
        list.__delslice__(self, start, stop)
        self.changed()

    def append(self, value):
        list.append(self, MutationObj.coerce(self._key, value))
        self.changed()

    def insert(self, idx, value):
        list.insert(self, idx, MutationObj.coerce(self._key, value))
        self.changed()

    def extend(self, values):
        list.extend(self, (MutationObj.coerce(self._key, v) for v in values))
        self.changed()

    def pop(self, *args, **kw):
        value = list.pop(self, *args, **kw)
        self.changed()
        return value

    def remove(self, value):
        list.remove(self, value)
        self.changed()

def JSONAlchemy(sqltype):
    """A type to encode/decode JSON on the fly

    sqltype is the string type for the underlying DB column.

    You can use it like:
    Column(JSONAlchemy(Text(600)))
    """
    class _JSONEncodedObj(JSONEncodedObj):
        impl = sqltype
    return MutationObj.as_mutable(_JSONEncodedObj)

Answer 3

I think the JSON example from the SQLAlchemy docs is also worth mentioning:

https://docs.sqlalchemy.org/en/13/core/custom_types.html#marshal-json-strings

However, I think it can be improved to be less strict regarding NULL and empty strings:

class JSONEncodedDict(TypeDecorator):
    impl = VARCHAR

    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return json.dumps(value, use_decimal=True)

    def process_result_value(self, value, dialect):
        if not value:
            return None
        return json.loads(value, use_decimal=True)

Answer 4

There is a recipe for this in the official documentation:

from sqlalchemy.types import TypeDecorator, VARCHAR
import json

class JSONEncodedDict(TypeDecorator):
    """Represents an immutable structure as a json-encoded string.

    Usage::

        JSONEncodedDict(255)

    """

    impl = VARCHAR

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = json.dumps(value)

        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = json.loads(value)
        return value

Answer 5

How about json.loads()?

>>> d= {"foo":1, "bar":[2,3]}
>>> s='{"foo":1, "bar":[2,3]}'
>>> import json
>>> json.loads(s) == d
True

Answer 6

Based on @snapshoe answer and to answer @Timmy's comment:

You can do it by using properties. Here is an example of a table:

class Providers(Base):
    __tablename__ = "providers"
    id = Column(
        Integer,
        Sequence('providers_id', optional=True),
        primary_key=True
    )
    name = Column(Unicode(40), index=True)
    _config = Column("config", Unicode(2048))

    @property
    def config(self):
        if not self._config:
            return {}
        return json.loads(self._config)

    @config.setter
    def config(self, value):
        self._config = json.dumps(value)

    def set_config(self, field, value):
        config = self.config
        config[field] = value
        self.config = config

    def get_config(self):
        if not self._config:
            return {}
        return json.loads(self._config)

    def unset_config(self, field):
        config = self.get_config()
        if field in config:
            del config[field]
            self.config = config

Now you can use it on a Providers() object:

>>> p = Providers()
>>> p.set_config("foo", "bar")
>>> p.get_config()
{"foo": "bar"}
>>> a.config
{u'foo': u'bar'}

I know this is an old Question maybe even dead, but I hope this could help someone.

Answer 7

This is what I came up with based on the two answers above.

import json

class JsonType(types.TypeDecorator):    

    impl = types.Unicode

    def process_bind_param(self, value, dialect):
        if value :
            return unicode(json.dumps(value))
        else:
            return {}

    def process_result_value(self, value, dialect):
        if value:
            return json.loads(value)
        else:
            return {}

Answer 8

As an update to the previous responses, which we've used with success so far. As of MySQL 5.7 and SQLAlchemy 1.1 you can use the native MySQL JSON data type, which gives you better performance and a whole range of operators for free.

It lets you to create virtual secondary indexes on JSON elements too.

But of course you will lock yourself into running your app on MySQL only when moving the logic into the database itself.