首页   注册   登录
V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
推荐学习书目
Learn Python the Hard Way
Python 学习手册
Python Cookbook
Python 基础教程
Python Sites
PyPI - Python Package Index
http://www.simple-is-better.com/
http://diveintopython.org/toc/index.html
Pocoo
值得关注的项目
PyPy
Celery
Jinja2
Read the Docs
gevent
pyenv
virtualenv
Stackless Python
Beautiful Soup
结巴中文分词
Green Unicorn
Sentry
Shovel
Pyflakes
pytest
Python 编程
pep8 Checker
Styles
PEP 8
Google Python Style Guide
Code Style from The Hitchhiker's Guide
华为云
V2EX  ›  Python

分享个 Sqlalchemy 的自定义字段类型, Mutation Tracking 问题

  •  
  •   fanhaipeng0403 · 5 天前 · 157 次点击

    JsonEncodeDict-demo-1.py

    import json
    
    # https://segmentfault.com/a/1190000004288061
    from sqlalchemy import Column, Integer, String
    from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.ext.mutable import Mutable
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.types import TypeDecorator, VARCHAR
    
    # ***************************
    engine = create_engine('sqlite:///./cnblogblog.db', echo=False)
    Base = declarative_base()
    DBSession = sessionmaker(bind=engine)
    
    
    class JSONEncodedDict(TypeDecorator):
        "Represents an immutable structure as a json-encoded string."
    
        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
    
    
    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()
    
    
    class MyDataClass1(Base):
        __tablename__ = 'my_data1'
        id = Column(Integer, primary_key=True)
        data = Column(MutableDict.as_mutable(JSONEncodedDict))
        name = Column(String(50))
    
    
    
    
    if __name__ == '__main__':
    
        Base.metadata.create_all(engine)
        session = DBSession()
    
        m1 = MyDataClass1(data={'value1': 'foo1'}, name='xiaohong')
        session.add(m1)
        session.commit()
    
        #######session 提交后,data 可以关联到 query
    
        m1.name = 'xiaolang'
        
        m1.data['value1'] = 'bar'#数据库的值,将再改变
        
        # assert m1 in session.dirty
        session.commit()
    
        # my_data= session.query(MyDataClass).filter_by(id=1).one()
        # a= my_data.data
        # print (type(a))
        # print (a)
        # print (a["value1"])
        #
        # my_data.data["value1"] = "foo2"
        #
        # session.commit()
    
    

    JsonEncodeDict-demo-2.py

    import json
    
    from sqlalchemy import Column, Integer, String
    # https://segmentfault.com/a/1190000004288061
    from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.types import TypeDecorator, VARCHAR
    
    # ***************************
    engine = create_engine('sqlite:///./cnblogblog.db', echo=False)
    Base = declarative_base()
    DBSession = sessionmaker(bind=engine)
    
    
    class JSONEncodedDict(TypeDecorator):
        "Represents an immutable structure as a json-encoded string."
    
        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
    
    
    #
    # 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()
    
    
    class MyDataClass2(Base):
        __tablename__ = 'my_data2'
        id = Column(Integer, primary_key=True)
        data = Column(JSONEncodedDict)
        name = Column(String(50))
    
    
    if __name__ == '__main__':
        Base.metadata.create_all(engine)
        session = DBSession()
    
        m1 = MyDataClass2(data={'value1': 'foo1'}, name='xiaohong')
        session.add(m1)
        session.commit()
    
        m1.name = 'xiaolang'
        #######session 提交后,name 关联到了 query
    
        # assert m1 in session.dirty
    
        #######session 提交后,data 将不再关联到 query(解决方法,按着 JsonEncodeDict-demo-1.py 来)
    
        m1.data['value1'] = 'bar'#数据库的值,不再改变
        session.commit()
        # assert m1 in session.dirty
    
        # my_data= session.query(MyDataClass).filter_by(id=1).one()
        # a= my_data.data
        # print (type(a))
        # print (a)
        # print (a["value1"])
        #
        # my_data.data["value1"] = "foo2"
        #
        # session.commit()
        ```
    
    目前尚无回复
    关于   ·   FAQ   ·   API   ·   我们的愿景   ·   广告投放   ·   感谢   ·   实用小工具   ·   2062 人在线   最高记录 3821   ·  
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.1 · 42ms · UTC 15:29 · PVG 23:29 · LAX 07:29 · JFK 10:29
    ♥ Do have faith in what you're doing.
    沪ICP备16043287号-1