SQLAlchemy JSON as blob/text

Question

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

Answer 1

Вы можете очень легко create your own type с SQLAlchemy

---

Для SQLAlchemy версии> = 0.7, проверить Yogesh's answer ниже

---

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.

Answer 2

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

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

Answer 3

Я думаю, что пример 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) 

Answer 4

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

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 5

sqlalchemy.types.MutableType устарел (v0.7 вперед), то documentation recommends с помощью 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) 

Answer 6

на основе @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'} 

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

Answer 7

Существует рецепт для этого в 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 8

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

Он также позволяет создавать virtual secondary indexes на элементах JSON.

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