SQLAlchemy JSON come blob/testo

Question

Sto memorizzando JSON come blob/testo in una colonna utilizzando MySQL. C'è un modo semplice per convertire questo in un dict usando python/SQLAlchemy?

Answer 1

si può facilmente create your own type con SQLAlchemy

---

Per le versioni SQLAlchemy> = 0,7, controlla Yogesh's answer seguito

---

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

Questo può essere utilizzato quando si sta definendo le tue tabelle (esempio usa elisir):

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) 

È anche possibile utilizzare un serializzatore json diverso per jsonpickle.

Answer 2

Che ne dici di json.loads()?

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

Answer 3

ritengo l'esempio JSON dalla documentazione SQLAlchemy anche la pena menzionare:

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

Ritengo tuttavia può essere migliorato per essere meno severe per quanto riguarda NULL e le stringhe vuote:

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

Questo è quello che mi è venuto in mente in base alle due risposte sopra.

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 è sconsigliata (v0.7 in poi), il documentation recommends utilizzando sqlalchemy.ext.mutable invece.

Ho trovato un Git gist di dbarnett che ho testato per il mio utilizzo. Ha funzionato bene finora, sia per il dizionario che per gli elenchi.

incolla di seguito per i posteri:

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

in base alla risposta @snapshoe e di rispondere @ commento di Timmy:

È possibile farlo utilizzando le proprietà. Ecco un esempio di una tabella:

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 

Ora è possibile utilizzarlo su un oggetto Providers():

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

So che questa è una vecchia questione forse anche morto, ma spero che questo potrebbe aiutare qualcuno .

Answer 7

C'è una ricetta per questo nella 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

come aggiornamento per le risposte precedenti, che abbiamo usato con successo finora. A partire da MySQL 5.7 e SQLAlchemy 1.1 è possibile utilizzare native MySQL JSON data type, che offre prestazioni migliori e un'intero range of operators gratuitamente.

Consente di creare anche virtual secondary indexes su elementi JSON.

Ma ovviamente ti bloccherai a eseguire l'app su MySQL solo quando sposti la logica nel database stesso.