SQLAlchemy JSON как blob/text

Ответ 1

Вы можете легко создать свой собственный тип с помощью SQLAlchemy

Для версий SQLAlchemy >= 0.7, ответ Yogesh ниже

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 {}

Это можно использовать, когда вы определяете свои таблицы (пример использует эликсир):

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)

Вы также можете использовать другой json serialiser для jsonpickle.

Ответ 2

Я думаю, что пример JSON из документов SQLAlchemy также стоит упомянуть:

http://www.sqlalchemy.org/docs/core/types.html#marshal-json-strings

Однако я думаю, что его можно улучшить, чтобы быть менее строгим относительно NULL и пустых строк:

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)

Ответ 3

Как насчет json.loads()?

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

Ответ 4

sqlalchemy.types.MutableType устарел (v0.7 вперед), Git gist dbarnett, который я тестировал для моего использования. До сих пор он работал хорошо, как для словарей, так и для списков.

Вставка ниже для потомков:

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)

Ответ 5

Существует рецепт для этого в официальной документации :

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

Ответ 6

Основываясь на ответе @snapshoe и отвечая на комментарий @Timmy:

Вы можете сделать это, используя свойства. Вот пример таблицы:

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

Теперь вы можете использовать его на объекте Providers():

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

Я знаю, что это старый вопрос, возможно, даже мертвый, но я надеюсь, что это может помочь кому-то.

Ответ 7

Это то, что я придумал, основываясь на двух ответах выше.

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 {}

Ответ 8

Как обновление предыдущих ответов, которые мы использовали с успехом. Начиная с MySQL 5.7 и SQLAlchemy 1.1 вы можете использовать диапазон операторов бесплатно.

Это позволяет создавать виртуальные вторичные индексы на элементах JSON.

Но, конечно, вы закроете себя при запуске своего приложения в MySQL только при перемещении логики в базу данных.