SQLAlchemy JSON als Blob/Text

Question

Ich speichere JSON als Blob/Text in einer Spalte mit MySQL. Gibt es eine einfache Möglichkeit, dies mit Python/SQLAlchemy in ein dict zu konvertieren?

Answer 1

Sie können sehr leicht create your own type mit SQLAlchemy

---

Für SQLAlchemy Versionen> = 0,7 Besuche Yogesh's answer unter

---

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

Dieses verwendet werden kann, wenn Sie definieren Ihre Tabellen (Beispiel verwendet Elixier):

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) 

Sie können auch einen anderen JSON-Serialiser für jsonpickle verwenden.

Answer 2

Wie wäre es mit json.loads()?

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

Answer 3

ich glaube, das JSON-Beispiel aus dem SQLAlchemy docs erwähnt auch wert ist:

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

Aber ich denke, es verbessert werden kann, weniger streng in Bezug auf NULL und leeren Strings sein:

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

Dies ist, was ich basierend auf den beiden oben genannten Antworten kam.

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 Weiter) veraltet ist, die documentation recommends mit sqlalchemy.ext.mutable statt.

Ich fand eine Git gist von dbarnett, die ich für meine Verwendung getestet habe. Es hat bisher gut funktioniert, sowohl für das Wörterbuch als auch für die Listen.

unter Einfügen für die Nachwelt:

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

Basierend auf @snapshoe Antwort und @ Timmys Kommentar zu beantworten:

Sie es Eigenschaften, indem Sie tun können. Hier ist ein Beispiel einer Tabelle:

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 

Jetzt können Sie es auf einem Providers() Objekt verwenden:

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

Ich weiß, dass dies eine alte Frage ist vielleicht sogar tot, aber ich hoffe, dass dies jemand helfen könnte .

Answer 7

Es ist ein Rezept für den in den 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

als Update zu den vorherigen Antworten, die wir bisher mit Erfolg verwendet haben. Ab MySQL 5.7 und SQLAlchemy 1.1 können Sie die native MySQL JSON data type verwenden, die Ihnen eine bessere Leistung und eine ganze range of operators kostenlos zur Verfügung stellt.

Damit können Sie auch virtual secondary indexes auf JSON-Elementen erstellen.

Aber natürlich sperren Sie Ihre Anwendung auf MySQL nur, wenn Sie die Logik in die Datenbank selbst verschieben.