SQLAlchemy JSON dưới dạng blob/text

Question

Tôi đang lưu trữ JSON dưới dạng blob/text trong một cột bằng cách sử dụng MySQL. Có một cách đơn giản để chuyển đổi này thành một dict bằng cách sử dụng python/SQLAlchemy?

Answer 1

Bạn có thể rất dễ dàng create your own type với SQLAlchemy

---

Đối với phiên bản SQLAlchemy> = 0.7, hãy kiểm tra Yogesh's answer dưới

---

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

này có thể được sử dụng khi bạn đang xác định bảng của bạn (ví dụ sử dụng elixir):

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) 

Bạn cũng có thể sử dụng trình tuần tự json khác với jsonpickle.

Answer 2

Làm thế nào về json.loads()?

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

Answer 3

Tôi nghĩ rằng ví dụ JSON từ các tài liệu SQLAlchemy cũng đáng nhắc đến:

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

Tuy nhiên, tôi nghĩ rằng nó có thể được cải thiện để được ít nghiêm ngặt về NULL và chuỗi rỗng:

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

Đây là những gì tôi nghĩ ra dựa trên hai câu trả lời ở trên.

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 đã bị phản đối (v0.7 trở đi), các documentation recommends sử dụng sqlalchemy.ext.mutable để thay thế.

Tôi đã tìm thấy Git gist bởi dbarnett mà tôi đã thử nghiệm để sử dụng. Nó đã làm việc tốt cho đến nay, cho cả từ điển và danh sách.

dán dưới đây cho hậu thế:

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

Dựa trên câu trả lời @snapshoe và để trả lời @ bình luận Timmy:

Bạn có thể làm điều đó bằng cách sử dụng tài sản. Dưới đây là một ví dụ về một bảng:

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 

Bây giờ bạn có thể sử dụng nó trên một đối tượng Providers():

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

Tôi biết đây là một câu hỏi cũ thậm chí có thể chết, nhưng tôi hy vọng điều này có thể giúp đỡ người khác .

Answer 7

Có một công thức cho việc này trong 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

Như một cập nhật cho các câu trả lời trước đó, mà chúng tôi đã sử dụng thành công cho đến nay. Kể từ MySQL 5.7 và SQLAlchemy 1.1, bạn có thể sử dụng native MySQL JSON data type, cung cấp cho bạn hiệu suất tốt hơn và toàn bộ miễn phí range of operators.

Nó cũng cho phép bạn tạo virtual secondary indexes trên các phần tử JSON.

Nhưng tất nhiên bạn sẽ tự khóa mình khi chạy ứng dụng của bạn trên MySQL chỉ khi di chuyển logic vào cơ sở dữ liệu.