使用sqlalchemy来读取的json格式的数据

sqlalchemy_json_column.py

#_*_ coding: utf-8 _*_

"""
Mutation Tracking
    http://sqlalchemy.readthedocs.org/en/rel_0_8/orm/extensions/mutable.html

JSONColumn
https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/JSONColumn

"""

from sqlalchemy.types import TypeDecorator, TEXT
import json

json_null = object()

class JSONEncodedDict(TypeDecorator):
    "Represents an immutable structure as a json-encoded string."

    impl = TEXT

    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
        
        
from sqlalchemy.ext.mutable import Mutable

class MutableDict(Mutable, dict):
    @classmethod
    def coerce(cls, key, value):
        "Convert plain dictionaries to MutableDict."

        if not isinstance(value, MutableDict):
            if isinstance(value, dict):
                return MutableDict(value)

            # this call will raise ValueError
            return Mutable.coerce(key, value)
        else:
            return value

    def __setitem__(self, key, value):
        "Detect dictionary set events and emit change events."

        dict.__setitem__(self, key, value)
        self.changed()

    def __delitem__(self, key):
        "Detect dictionary del events and emit change events."

        dict.__delitem__(self, key)
        self.changed()
        
        
MutableDict.associate_with(JSONEncodedDict)


if __name__ == '__main__':
    from sqlalchemy import *
    engine = create_engine('sqlite://', echo=True)

    m = MetaData(engine)

    t1 = Table('table', m,
        Column('id', Integer, primary_key=True),
        Column('data', JSONEncodedDict)
    )

    m.create_all()


    t1.insert().execute(
            #{'data':"a string", id:1},    # 会导致json load失败 
            {'data':{"a":{'c':1, 'd':2}}, id:2},
            {'data':None, id:3},
    )

    print t1.select().where(t1.c.data=={}).execute().fetchall()
    print t1.select().where(t1.c.data!={}).execute().fetchall()
    print t1.select().where(t1.c.data==None).execute().fetchall()
    
    
        class T(object):
        pass

    from sqlalchemy.orm import mapper

    mapper(T, t1)

    from sqlalchemy.orm import Session

    sess = Session()
    t = T()
    t.data = {'b': 1}
    sess.add(t)
    sess.commit()
    print t.data['b']
    print sess.query(T).filter(T.data=="{'b':1}").all()
    tlist = sess.query(T).filter(T.data=={'b':1}).all()
    for t in tlist: print t.data.keys()

    # NOTE 用like的方式好像无法生效
    print sess.query(T).filter(T.data.like('%b%')).first()

    t2 = sess.query(T).first()
    print t2.data
    # 直接改dict里面的dict无法检查到changed. 像django的ssession, 主动changed一下
    t2.data['a']['c'] = 2
    t2.data.changed()
    #t2.data['a'] = 1
    print sess.dirty