SQLAlchemy JSON как blob / text - PullRequest
       26

SQLAlchemy JSON как blob / text

15 голосов
/ 28 октября 2010

Я храню JSON как blob / text в столбце, используя MySQL.Есть ли простой способ преобразовать это в dict, используя python / SQLAlchemy?

Ответы [ 8 ]

14 голосов
/ 29 октября 2010

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


Для версий SQLAlchemy> = 0,7 см. ответ Йогеша ниже


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 для jsonpickle.

8 голосов
/ 22 апреля 2011

Я думаю, что пример 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)
7 голосов
/ 06 января 2015

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

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 голосов
/ 29 августа 2014

sqlalchemy.types.MutableType устарело (начиная с версии 0.7), документация рекомендует с использованием sqlalchemy.ext.mutable.

Я нашел 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)
6 голосов
/ 28 октября 2010

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

>>> d= {"foo":1, "bar":[2,3]}
>>> s='{"foo":1, "bar":[2,3]}'
>>> import json
>>> json.loads(s) == d
True
2 голосов
/ 28 октября 2014

На основе ответа @snapshoe и ответа на комментарий @ Тимми:

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

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

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

1 голос
/ 10 июня 2016

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

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

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

1 голос
/ 11 июня 2013

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

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 {}
Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...