Skip to content

Instantly share code, notes, and snippets.

@vfr292

vfr292/review_postgres_sql-alchmey_json_field_properties.py

Last active June 28, 2018 15:53
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save vfr292/41530bfb56778ffc26fe53c605f40feb to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save vfr292/41530bfb56778ffc26fe53c605f40feb to your computer and use it in GitHub Desktop.
Download ZIP
A review of querying/searching and mutability tracking on Postgres SQL JSON and JSONB field implementations using SQLAlchemy and Sqlalchemy-Json
Raw
review_postgres_sql-alchmey_json_field_properties.py
import json
from contextlib import contextmanager
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.types import JSON
from sqlalchemy.dialects.postgresql import JSON as pg_JSON, JSONB as pg_JSONB
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy_json import NestedMutableJson, MutableJson, NestedMutable as sqlalchemy_json_NestedMutable
from car_listing_flask_app.config import TestingConfig
'''
The file prints whether or not the json field implementation had the desired behavior.
'''
Base = declarative_base()
class JsonData(Base):
__tablename__ = 'json_datas'
id = Column(Integer, primary_key=True)
basic_json = Column(JSON)
pg_jsonb = Column(pg_JSONB)
jsonb_mutable_dict = Column(MutableDict.as_mutable(pg_JSONB))
nested_mutable_json = Column(NestedMutableJson)
pg_jsonb_manual = Column(pg_JSON)
nested_mutable_as_mutable = Column(sqlalchemy_json_NestedMutable.as_mutable(pg_JSONB))
@classmethod
def get_json_field_names(cls):
return ['basic_json', 'pg_jsonb', 'jsonb_mutable_dict', 'nested_mutable_json', 'pg_jsonb_manual', 'nested_mutable_as_mutable']
@classmethod
def initialize_instance_with_value(cls, value):
return cls(basic_json = value, pg_jsonb = value, jsonb_mutable_dict = value, nested_mutable_json = value, pg_jsonb_manual = json.dumps(value), nested_mutable_as_mutable = value)
def update_json_keys_values(self, key, value):
for f_name in self.get_json_field_names():
if f_name == 'pg_jsonb_manual':
data = json.loads(self.pg_jsonb_manual)
data['key'] = value
self.pg_jsonb_manual = json.dumps(data)
else:
getattr(self, f_name)[key] = value
def append_value_to_json_dicts(self, key, value):
for f_name in self.get_json_field_names():
if f_name == 'pg_jsonb_manual':
data = json.loads(self.pg_jsonb_manual)
data['key'].append(value)
self.pg_jsonb_manual = json.dumps(data)
else:
getattr(self, f_name)[key].append(value)
def append_value_to_json_nested_dicts(self, key1, key2, value):
for f_name in self.get_json_field_names():
if f_name == 'pg_jsonb_manual':
data = json.loads(self.pg_jsonb_manual)
data['key1']['key2'].append(value)
self.pg_jsonb_manual = json.dumps(data)
else:
getattr(self, f_name)[key1][key2].append(value)
@contextmanager
def setup_teardown():
engine = create_engine(TestingConfig.SQLALCHEMY_DATABASE_URI)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
yield (session, engine)
session.close()
Base.metadata.drop_all(engine)
def test_basic_key_value_insert():
#tests adding a dict to a json field
with setup_teardown() as i:
session, engine = i
value = dict(key = ['test'])
jd = JsonData.initialize_instance_with_value(value)
session.add(jd)
session.commit()
for f_name in JsonData.get_json_field_names():
f_val = getattr(jd, f_name)
try:
assert f_val == value or (f_name == 'pg_jsonb_manual' and json.loads(f_val) == value), '{} did not equal value, but {}'.format(f_name, f_val)
print('test_basic_key_value_insert for {}'.format(f_name))
except AssertionError as e:
print('{} did not equal {}, but {}'.format(f_name, value, f_val))
def test_search_by_key_value():
#tests querying the field for eqaulity with the dict's value
with setup_teardown() as i:
session, engine = i
value = dict(key = 'test')
jd = JsonData.initialize_instance_with_value(value)
session.add(jd)
session.commit()
# print('ECHO ON')
# engine.echo = True
for f_name in JsonData.get_json_field_names():
try:
if f_name == 'pg_jsonb_manual':
jd = session.query(JsonData).filter(
getattr(JsonData, f_name)['key'].astext == json.dumps('test')
).one()
else:
jd = session.query(JsonData).filter(
getattr(JsonData, f_name)['key'].astext == 'test'
).one()
print('test_search_by_key_value for {}'.format(f_name))
except AttributeError as e:
print('FAIL {} could not search for key. AttributeError: {}'.format(f_name, e))
except NotImplementedError as e:
print('FAIL {} could not search for key. NotImplementedError: {}'.format(f_name, e))
except NoResultFound as e:
print('FAIL {} could not search for key. NoResultFound: {}'.format(f_name, e))
# engine.echo = False
def test_search_by_key_value_explicit():
#I was getting a NotImplementedError with nested_mutable_json and wanted to do a more explicit test.
with setup_teardown() as i:
session, engine = i
value = dict(key = 'test')
jd = JsonData(nested_mutable_json = value, jsonb_mutable_dict = value)
session.add(jd)
session.commit()
print(jd.nested_mutable_json) #looks OK
print(jd.nested_mutable_json['key']) #looks OK
jd = session.query(JsonData).filter(
JsonData.jsonb_mutable_dict['key'].astext == 'test'
).one() #worked OK
print('instance retrieved with jsonb_mutable_dict')
jd = session.query(JsonData).filter(
JsonData.nested_mutable_json['key'].astext == 'test'
).one()
print('test_search_by_key_value for nested_mutable_json')
def test_change_value_of_json_dicts():
#tests changing the value of a dictionary key
with setup_teardown() as i:
session, engine = i
value = dict(key = 'test')
jd = JsonData.initialize_instance_with_value(value)
session.add(jd)
session.commit()
jd.update_json_keys_values('key', 'new test')
session.commit()
expected_value = dict(key = 'new test')
for f_name in JsonData.get_json_field_names():
f_val = getattr(jd, f_name)
try:
assert f_val == expected_value or (f_name == 'pg_jsonb_manual' and json.loads(f_val) == expected_value)
print('test_value_append for {}'.format(f_name))
except AssertionError as e:
print('FAIL {} did not equal {}, but {}'.format(f_name, value, f_val))
def test_append_value_to_json_dicts():
with setup_teardown() as i:
session, engine = i
value = dict(key = ['test'])
jd = JsonData.initialize_instance_with_value(value)
session.add(jd)
session.commit()
jd.append_value_to_json_dicts('key', 'new test')
session.commit()
expected_value = dict(key = ['test', 'new test'])
for f_name in JsonData.get_json_field_names():
f_val = getattr(jd, f_name)
try:
assert f_val == expected_value or (f_name == 'pg_jsonb_manual' and json.loads(f_val) == expected_value)
print('test_value_append for {}'.format(f_name))
except AssertionError as e:
print('FAIL {} did not equal {}, but {}'.format(f_name, value, f_val))
def test_append_value_to_json_nested_dicts():
with setup_teardown() as i:
session, engine = i
value = dict(key1 = dict(key2 = ['test']))
jd = JsonData.initialize_instance_with_value(value)
session.add(jd)
session.commit()
jd.append_value_to_json_nested_dicts('key1', 'key2', 'new test')
session.commit()
expected_value = dict(key1 = dict(key2 = ['test', 'new test']))
for f_name in JsonData.get_json_field_names():
f_val = getattr(jd, f_name)
try:
assert f_val == expected_value or (f_name == 'pg_jsonb_manual' and json.loads(f_val) == expected_value)
print('test_append_value_to_json_nested_dicts for {}'.format(f_name))
except AssertionError as e:
print('FAIL {} did not equal {}, but {}'.format(f_name, value, f_val))
if __name__ == "__main__":
print('test_basic_key_value_insert'.upper())
test_basic_key_value_insert()
print('test_search_by_key_value'.upper())
# test_search_by_key_value_explicit()
test_search_by_key_value()
print('test_change_value_of_json_dicts'.upper())
test_change_value_of_json_dicts()
print('test_append_value_to_json_dicts'.upper())
test_append_value_to_json_dicts()
print(test_append_value_to_json_nested_dicts)
test_append_value_to_json_nested_dicts()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment