# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from sqlalchemy import Column, Integer, create_engine
from sqlalchemy.orm import Session
from sqlalchemy.orm.attributes import flag_modified
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.postgresql import JSONB
Base = declarative_base()
class EntityA(Base):
__tablename__ = 'entity_a'
a_id = Column(Integer, primary_key=True)
a_value = Column(JSONB, nullable=False)
class EntityB(Base):
__tablename__ = 'entity_b'
b_id = Column(Integer, primary_key=True)
_b_value = Column(JSONB, nullable=False)
@hybrid_property
def b_value(self):
return self._b_value
@b_value.setter
def b_value(self, value):
self._b_value = value
class EntityC(Base):
__tablename__ = 'entity_c'
c_id = Column(Integer, primary_key=True)
_c_value = Column(JSONB, nullable=False)
@hybrid_property
def c_value(self):
return self._c_value
@c_value.setter
def c_value(self, value):
self._c_value = value
flag_modified(self, '_c_value')
if __name__ == '__main__':
engine =
create_engine('postgresql://test:test123@127.0.0.1/testing', echo=False)
Base.metadata.create_all(engine)
session = Session(engine)
# testing A, without hybrid property
a_test = EntityA()
a_test.a_value = dict(hello='world', test=1, enabled=True)
session.add(a_test)
session.commit()
print('EntityA, after insert:', a_test.a_value)
json_obj_a = a_test.a_value
json_obj_a['hello'] = 'foo'
a_test.a_value = json_obj_a
session.commit()
# assert a_test.a_value.get('hello') == 'foo' # error
assert a_test.a_value.get('hello') == 'world' # works
print('EntityA, after update:', a_test.a_value)
# testing B, with hybrid property and no flag_modified
b_test = EntityB()
b_test.b_value = dict(hello='world', test=1, enabled=True)
session.add(b_test)
session.commit()
print('EntityB, after insert:', b_test.b_value)
json_obj_b = b_test.b_value
json_obj_b['hello'] = 'bar'
b_test.b_value = json_obj_b
session.commit()
# assert b_test.b_value.get('hello') == 'bar' # error
assert b_test.b_value.get('hello') == 'world' # works
print('EntityB, after update:', b_test.b_value)
# testing C, with hybrid property and flag_modified
c_test = EntityC()
c_test.c_value = dict(hello='world', test=1, enabled=True)
session.add(c_test)
session.commit()
print('EntityC, after insert:', c_test.c_value)
json_obj_c = c_test.c_value
json_obj_c['hello'] = 'baz'
c_test.c_value = json_obj_c
session.commit()
assert c_test.c_value.get('hello') == 'baz' # works
# assert c_test.c_value.get('hello') == 'world' # error
print('EntityC, after update:', c_test.c_value)
# end test
session.close()
Base.metadata.drop_all(engine)
extra data, if needed:
$ pip freeze
psycopg2==2.6.1
SQLAlchemy==1.0.8
$ postgres --version
postgres (PostgreSQL) 9.4.4
$ python --version
Python 2.7.10
$ uname -a
Linux marrow.polluxnet 4.2.3-1-ARCH #1 SMP PREEMPT Sat Oct 3
18:52:50 CEST 2015 x86_64 GNU/Linux
thanks a lot, if any other information is needed, please let me know.
best regards,
richard.
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
<<attachment: richard.vcf>>
