Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP

Loading…

Issue 2581 -- support for postgresql json type #50

Merged
merged 6 commits into from

2 participants

@nathan-rice

Very simple, based on the HSTORE data type.

I added support on the type itself for custom serializer/deserializer, and a different json module can be monkey patched in and will work properly (assuming it is api-compatible with python's json module, which most of them are).

nathan-rice added some commits
@nathan-rice nathan-rice sqlalchemy/dialects/postgresql/__init__.py:
- Added import references to JSON class

 sqlalchemy/dialects/postgresql/base.py:
 - Added visitor method for JSON class

 sqlalchemy/dialects/postgresql/pgjson (new):
 - JSON class, supports automatic serialization and deserialization of json data, as well as basic json operators.
64288c7
@nathan-rice nathan-rice Merge branch 'rel_0_9' of https://github.com/nathan-rice/sqlalchemy i…
…nto rel_0_9
ba73d61
lib/sqlalchemy/dialects/postgresql/pgjson.py
((4 lines not shown))
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+
+import json
+
+from .base import ARRAY, ischema_names
+from ... import types as sqltypes
+from ...sql import functions as sqlfunc
+from ...sql.operators import custom_op
+from ... import util
+
+__all__ = ('JSON', 'json')
+
+
+class JSON(sqltypes.TypeEngine):
+ """Represent the Postgresql HSTORE type.
@zzzeek Owner
zzzeek added a note

change this to JSON ;)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
lib/sqlalchemy/dialects/postgresql/pgjson.py
((29 lines not shown))
+ conn.execute(
+ data_table.insert(),
+ data = {"key1": "value1", "key2": "value2"}
+ )
+
+ :class:`.JSON` provides two operations:
+
+ * Index operations::
+
+ data_table.c.data['some key'] == 'some value'
+
+ * Path Index operations::
+
+ data_table.c.data.get_path('{key_1, key_2, ..., key_n}']
+
+ Please be aware that when used with the SQL Alchemy ORM, you will need to
@zzzeek Owner
zzzeek added a note

SQLAlchemy, seems like a misbehaving spellchecker

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
@zzzeek zzzeek commented on the diff
lib/sqlalchemy/dialects/postgresql/pgjson.py
((91 lines not shown))
+ def process(value):
+ return self.json_serializer(value).encode(encoding)
+ else:
+ def process(value):
+ return self.json_serializer(value)
+ return process
+
+ def result_processor(self, dialect, coltype):
+ if util.py2k:
+ encoding = dialect.encoding
+ def process(value):
+ return self.json_deserializer(value.decode(encoding))
+ else:
+ def process(value):
+ return self.json_deserializer(value)
+ return process
@zzzeek Owner
zzzeek added a note

Thanks for this note. It turns out that psycopg2's json support is pretty recent (up to date ubuntu packages don't have it), and it is on by default, so this code will actually error tests if psycopg2 is up to date. I'll see what I can do about this.

There is a problem having a use_native_json kwarg. Specifically, in order to not use native json with psycopg2 >= 2.5, you have to monkey-patch the module. I could check on dialect init to see if the module has been monkey-patched and revert if necessary, but I think it would be better to just pass on letting the user turn off json processing.

@zzzeek Owner
zzzeek added a note

that's not a problem, we don't need a "use_native_json" kwarg then - just use psycopg2's facility when it is detected (though we have to figure out a decent way to test the non-psycopg2 version considering that the tests usually run just on psycopg2...)

@zzzeek Owner
zzzeek added a note

ok we maybe can test it if we detect psycopg2 using their Json object with a null-dumper, but don't worry about that part, that's kind of intricate and I'll figure out how that might work. if we can just get tests to pass for now that's enough for the PR.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
test/dialect/postgresql/test_types.py
((105 lines not shown))
+ data = testing.db.execute(
+ select([self.tables.data_table.c.data]).
+ order_by(self.tables.data_table.c.name)
+ ).fetchall()
+ eq_([d for d, in data], compare)
+
+ def _test_insert(self, engine):
+ engine.execute(
+ self.tables.data_table.insert(),
+ {'name': 'r1', 'data': {"k1": "r1v1", "k2": "r1v2"}}
+ )
+ self._assert_data([{"k1": "r1v1", "k2": "r1v2"}])
+
+ def _non_native_engine(self):
+ if testing.against("postgresql+psycopg2"):
+ engine = engines.testing_engine(options=dict(use_native_hstore=False))
@zzzeek Owner
zzzeek added a note

yeah this would have to be "use_native_json=False" once that flag is implemented

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
@zzzeek
Owner

looks very nice, if we can add the psycopg2 support and fix a few typos we'll be in good shape :)

nathan-rice added some commits
@nathan-rice nathan-rice sqlalchemy/dialects/postgresql/pgjson:
 - Fixed reference to HSTORE
 - Corrected spelling of SQLAlchemy

 sqlalchemy/dialects/postgresql/psycopg2:
 - Added psycopg2 specific wrapper type for JSON which uses inherent json deserialization facilities
 - Added code to detect and utilize the JSON wrapper if psycopg2 >= 2.5

test/dialect/postgresql/test_types:
- removed reference to use_native_hstore
f285b35
@nathan-rice nathan-rice sqlalchemy/dialects/postgresql/psycopg2:
 - Removed unneeded import of psycopg2.extensions
059039e
@nathan-rice nathan-rice sqlalchemy/dialects/postgresql/pgjson:
 - Added support for additional operators
 - Made return as json default (rather than text)
015c73c
@nathan-rice nathan-rice sqlalchemy/dialects/postgresql/pgjson:
 - Updated documentation for JSON class
c64b7aa
@nathan-rice

This should be good to go now.

@zzzeek zzzeek commented on the diff
lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -363,6 +377,7 @@ def initialize(self, connection):
self._has_native_hstore = self.use_native_hstore and \
self._hstore_oids(connection.connection) \
is not None
+ self._has_native_json = self.psycopg2_version >= (2, 5)
@zzzeek Owner
zzzeek added a note

nailed it! nice.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
@zzzeek zzzeek commented on the diff
test/dialect/postgresql/test_types.py
((79 lines not shown))
+
+ def test_where_path_as_text(self):
+ self._test_where(
+ self.jsoncol.get_path_as_text('{"foo", 1}') == None,
+ "(test_table.test_column #>> %(test_column_1)s) IS NULL"
+ )
+
+ def test_cols_get(self):
+ self._test_cols(
+ self.jsoncol['foo'],
+ "test_table.test_column -> %(test_column_1)s AS anon_1",
+ True
+ )
+
+
+class JSONRoundTripTest(fixtures.TablesTest):
@zzzeek Owner
zzzeek added a note

great job on this

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
@zzzeek zzzeek commented on the diff
lib/sqlalchemy/dialects/postgresql/pgjson.py
((74 lines not shown))
+ def __getitem__(self, other):
+ """Text expression. Get the value at a given key."""
+ # I'm choosing to return text here so the result can be cast,
+ # compared with strings, etc.
+ #
+ # The only downside to this is that you cannot dereference more
+ # than one level deep in json structures, though comparator
+ # support for multi-level dereference is lacking anyhow.
+ return self.expr.op('->', precedence=5)(other)
+
+ def get_item_as_text(self, other):
+ """Text expression. Get the value at the given key as text. Use
+ this when you need to cast the type of the returned value."""
+ return self.expr.op('->>', precedence=5)(other)
+
+ def get_path(self, other):
@zzzeek Owner
zzzeek added a note

im thinking of streamlining the operators here. Looking at http://www.postgresql.org/docs/9.3/static/functions-json.html it seems like our operators are: getitem, getarray, and then both have the option to be "text".

so how about this:

json_col['some_element']    # get the element
json_col.astext['some_element']    # get the element as text
json_col[('a', 'b', 2)]  # get by path
json_col.astext[('a', 'b', 2)] # get by path as text

seems like also these ops are only in pg 9.3....ill add some qualifiers to the tests and I also want to add some round trips for these. going to install 9.3 now...

That is a much nicer interface, for sure.

What all are you planning on coding? I'm happy to code up all the stuff you mentioned, I just don't want to duplicate work.

@zzzeek Owner
zzzeek added a note

i think you're done! im just going to try to test this stuff out and familiarize, make that change, add a few more tests. you've laid out all the groundwork here so thanks!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
@zzzeek zzzeek merged commit c64b7aa into zzzeek:master
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Commits on Dec 9, 2013
  1. @nathan-rice

    sqlalchemy/dialects/postgresql/__init__.py:

    nathan-rice authored
    - Added import references to JSON class
    
     sqlalchemy/dialects/postgresql/base.py:
     - Added visitor method for JSON class
    
     sqlalchemy/dialects/postgresql/pgjson (new):
     - JSON class, supports automatic serialization and deserialization of json data, as well as basic json operators.
  2. @nathan-rice
Commits on Dec 10, 2013
  1. @nathan-rice

    sqlalchemy/dialects/postgresql/pgjson:

    nathan-rice authored
     - Fixed reference to HSTORE
     - Corrected spelling of SQLAlchemy
    
     sqlalchemy/dialects/postgresql/psycopg2:
     - Added psycopg2 specific wrapper type for JSON which uses inherent json deserialization facilities
     - Added code to detect and utilize the JSON wrapper if psycopg2 >= 2.5
    
    test/dialect/postgresql/test_types:
    - removed reference to use_native_hstore
  2. @nathan-rice

    sqlalchemy/dialects/postgresql/psycopg2:

    nathan-rice authored
     - Removed unneeded import of psycopg2.extensions
Commits on Dec 11, 2013
  1. @nathan-rice

    sqlalchemy/dialects/postgresql/pgjson:

    nathan-rice authored
     - Added support for additional operators
     - Made return as json default (rather than text)
  2. @nathan-rice

    sqlalchemy/dialects/postgresql/pgjson:

    nathan-rice authored
     - Updated documentation for JSON class
This page is out of date. Refresh to see the latest.
View
3  lib/sqlalchemy/dialects/postgresql/__init__.py
@@ -14,6 +14,7 @@
DATE, BYTEA, BOOLEAN, INTERVAL, ARRAY, ENUM, dialect, array, Any, All
from .constraints import ExcludeConstraint
from .hstore import HSTORE, hstore
+from .pgjson import JSON
from .ranges import INT4RANGE, INT8RANGE, NUMRANGE, DATERANGE, TSRANGE, \
TSTZRANGE
@@ -23,5 +24,5 @@
'DOUBLE_PRECISION', 'TIMESTAMP', 'TIME', 'DATE', 'BYTEA', 'BOOLEAN',
'INTERVAL', 'ARRAY', 'ENUM', 'dialect', 'Any', 'All', 'array', 'HSTORE',
'hstore', 'INT4RANGE', 'INT8RANGE', 'NUMRANGE', 'DATERANGE',
- 'TSRANGE', 'TSTZRANGE'
+ 'TSRANGE', 'TSTZRANGE', 'json', 'JSON'
)
View
3  lib/sqlalchemy/dialects/postgresql/base.py
@@ -1187,6 +1187,9 @@ def visit_BIGINT(self, type_):
def visit_HSTORE(self, type_):
return "HSTORE"
+ def visit_JSON(self, type_):
+ return "JSON"
+
def visit_INT4RANGE(self, type_):
return "INT4RANGE"
View
128 lib/sqlalchemy/dialects/postgresql/pgjson.py
@@ -0,0 +1,128 @@
+# postgresql/json.py
+# Copyright (C) 2005-2013 the SQLAlchemy authors and contributors <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+
+import json
+
+from .base import ARRAY, ischema_names
+from ... import types as sqltypes
+from ...sql import functions as sqlfunc
+from ...sql.operators import custom_op
+from ... import util
+
+__all__ = ('JSON', 'json')
+
+
+class JSON(sqltypes.TypeEngine):
+ """Represent the Postgresql JSON type.
+
+ The :class:`.JSON` type stores arbitrary JSON format data, e.g.::
+
+ data_table = Table('data_table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', JSON)
+ )
+
+ with engine.connect() as conn:
+ conn.execute(
+ data_table.insert(),
+ data = {"key1": "value1", "key2": "value2"}
+ )
+
+ :class:`.JSON` provides several operations:
+
+ * Index operations::
+
+ data_table.c.data['some key']
+
+ * Index operations returning text (required for text comparison or casting)::
+
+ data_table.c.data.get_item_as_text('some key') == 'some value'
+
+ * Path index operations::
+
+ data_table.c.data.get_path("{key_1, key_2, ..., key_n}")
+
+ * Path index operations returning text (required for text comparison or casting)::
+
+ data_table.c.data.get_path("{key_1, key_2, ..., key_n}") == 'some value'
+
+ Please be aware that when used with the SQLAlchemy ORM, you will need to
+ replace the JSON object present on an attribute with a new object in order
+ for any changes to be properly persisted.
+
+ .. versionadded:: 0.9
+ """
+
+ __visit_name__ = 'JSON'
+
+ def __init__(self, json_serializer=None, json_deserializer=None):
+ if json_serializer:
+ self.json_serializer = json_serializer
+ else:
+ self.json_serializer = json.dumps
+ if json_deserializer:
+ self.json_deserializer = json_deserializer
+ else:
+ self.json_deserializer = json.loads
+
+ class comparator_factory(sqltypes.Concatenable.Comparator):
+ """Define comparison operations for :class:`.JSON`."""
+
+ def __getitem__(self, other):
+ """Text expression. Get the value at a given key."""
+ # I'm choosing to return text here so the result can be cast,
+ # compared with strings, etc.
+ #
+ # The only downside to this is that you cannot dereference more
+ # than one level deep in json structures, though comparator
+ # support for multi-level dereference is lacking anyhow.
+ return self.expr.op('->', precedence=5)(other)
+
+ def get_item_as_text(self, other):
+ """Text expression. Get the value at the given key as text. Use
+ this when you need to cast the type of the returned value."""
+ return self.expr.op('->>', precedence=5)(other)
+
+ def get_path(self, other):
@zzzeek Owner
zzzeek added a note

im thinking of streamlining the operators here. Looking at http://www.postgresql.org/docs/9.3/static/functions-json.html it seems like our operators are: getitem, getarray, and then both have the option to be "text".

so how about this:

json_col['some_element']    # get the element
json_col.astext['some_element']    # get the element as text
json_col[('a', 'b', 2)]  # get by path
json_col.astext[('a', 'b', 2)] # get by path as text

seems like also these ops are only in pg 9.3....ill add some qualifiers to the tests and I also want to add some round trips for these. going to install 9.3 now...

That is a much nicer interface, for sure.

What all are you planning on coding? I'm happy to code up all the stuff you mentioned, I just don't want to duplicate work.

@zzzeek Owner
zzzeek added a note

i think you're done! im just going to try to test this stuff out and familiarize, make that change, add a few more tests. you've laid out all the groundwork here so thanks!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
+ """Text expression. Get the value at a given path. Paths are of
+ the form {key_1, key_2, ..., key_n}."""
+ return self.expr.op('#>', precedence=5)(other)
+
+ def get_path_as_text(self, other):
+ """Text expression. Get the value at a given path, as text.
+ Paths are of the form {key_1, key_2, ..., key_n}. Use this when
+ you need to cast the type of the returned value."""
+ return self.expr.op('#>>', precedence=5)(other)
+
+ def _adapt_expression(self, op, other_comparator):
+ if isinstance(op, custom_op):
+ if op.opstring == '->':
+ return op, sqltypes.Text
+ return sqltypes.Concatenable.Comparator.\
+ _adapt_expression(self, op, other_comparator)
+
+ def bind_processor(self, dialect):
+ if util.py2k:
+ encoding = dialect.encoding
+ def process(value):
+ return self.json_serializer(value).encode(encoding)
+ else:
+ def process(value):
+ return self.json_serializer(value)
+ return process
+
+ def result_processor(self, dialect, coltype):
+ if util.py2k:
+ encoding = dialect.encoding
+ def process(value):
+ return self.json_deserializer(value.decode(encoding))
+ else:
+ def process(value):
+ return self.json_deserializer(value)
+ return process
@zzzeek Owner
zzzeek added a note

Thanks for this note. It turns out that psycopg2's json support is pretty recent (up to date ubuntu packages don't have it), and it is on by default, so this code will actually error tests if psycopg2 is up to date. I'll see what I can do about this.

There is a problem having a use_native_json kwarg. Specifically, in order to not use native json with psycopg2 >= 2.5, you have to monkey-patch the module. I could check on dialect init to see if the module has been monkey-patched and revert if necessary, but I think it would be better to just pass on letting the user turn off json processing.

@zzzeek Owner
zzzeek added a note

that's not a problem, we don't need a "use_native_json" kwarg then - just use psycopg2's facility when it is detected (though we have to figure out a decent way to test the non-psycopg2 version considering that the tests usually run just on psycopg2...)

@zzzeek Owner
zzzeek added a note

ok we maybe can test it if we detect psycopg2 using their Json object with a null-dumper, but don't worry about that part, that's kind of intricate and I'll figure out how that might work. if we can just get tests to pass for now that's enough for the PR.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
+
+
+ischema_names['json'] = JSON
View
15 lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -179,6 +179,7 @@
ENUM, ARRAY, _DECIMAL_TYPES, _FLOAT_TYPES,\
_INT_TYPES
from .hstore import HSTORE
+from .pgjson import JSON
logger = logging.getLogger('sqlalchemy.dialects.postgresql')
@@ -243,6 +244,17 @@ def result_processor(self, dialect, coltype):
else:
return super(_PGHStore, self).result_processor(dialect, coltype)
+
+class _PGJSON(JSON):
+ # I've omitted the bind processor here because the method of serializing
+ # involves registering specific types to auto-serialize, and the adapter
+ # just a thin wrapper over json.dumps.
+ def result_processor(self, dialect, coltype):
+ if dialect._has_native_json:
+ return None
+ else:
+ return super(_PGJSON, self).result_processor(dialect, coltype)
+
# When we're handed literal SQL, ensure it's a SELECT-query. Since
# 8.3, combining cursors and "FOR UPDATE" has been fine.
SERVER_SIDE_CURSOR_RE = re.compile(
@@ -327,6 +339,7 @@ class PGDialect_psycopg2(PGDialect):
psycopg2_version = (0, 0)
_has_native_hstore = False
+ _has_native_json = False
colspecs = util.update_copy(
PGDialect.colspecs,
@@ -336,6 +349,7 @@ class PGDialect_psycopg2(PGDialect):
sqltypes.Enum: _PGEnum, # needs force_unicode
ARRAY: _PGArray, # needs force_unicode
HSTORE: _PGHStore,
+ JSON: _PGJSON
}
)
@@ -363,6 +377,7 @@ def initialize(self, connection):
self._has_native_hstore = self.use_native_hstore and \
self._hstore_oids(connection.connection) \
is not None
+ self._has_native_json = self.psycopg2_version >= (2, 5)
@zzzeek Owner
zzzeek added a note

nailed it! nice.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
@classmethod
def dbapi(cls):
View
191 test/dialect/postgresql/test_types.py
@@ -15,7 +15,8 @@
from sqlalchemy import exc, schema, types
from sqlalchemy.dialects.postgresql import base as postgresql
from sqlalchemy.dialects.postgresql import HSTORE, hstore, array, \
- INT4RANGE, INT8RANGE, NUMRANGE, DATERANGE, TSRANGE, TSTZRANGE
+ INT4RANGE, INT8RANGE, NUMRANGE, DATERANGE, TSRANGE, TSTZRANGE, \
+ JSON
import decimal
from sqlalchemy import util
from sqlalchemy.testing.util import round_decimal
@@ -1651,3 +1652,191 @@ def _data_str(self):
def _data_obj(self):
return self.extras.DateTimeTZRange(*self.tstzs())
+
+
+class JSONTest(fixtures.TestBase):
+ def _assert_sql(self, construct, expected):
+ dialect = postgresql.dialect()
+ compiled = str(construct.compile(dialect=dialect))
+ compiled = re.sub(r'\s+', ' ', compiled)
+ expected = re.sub(r'\s+', ' ', expected)
+ eq_(compiled, expected)
+
+ def setup(self):
+ metadata = MetaData()
+ self.test_table = Table('test_table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('test_column', JSON)
+ )
+ self.jsoncol = self.test_table.c.test_column
+
+ def _test_where(self, whereclause, expected):
+ stmt = select([self.test_table]).where(whereclause)
+ self._assert_sql(
+ stmt,
+ "SELECT test_table.id, test_table.test_column FROM test_table "
+ "WHERE %s" % expected
+ )
+
+ def _test_cols(self, colclause, expected, from_=True):
+ stmt = select([colclause])
+ self._assert_sql(
+ stmt,
+ (
+ "SELECT %s" +
+ (" FROM test_table" if from_ else "")
+ ) % expected
+ )
+
+ def test_bind_serialize_default(self):
+ from sqlalchemy.engine import default
+
+ dialect = default.DefaultDialect()
+ proc = self.test_table.c.test_column.type._cached_bind_processor(dialect)
+ eq_(
+ proc({"A": [1, 2, 3, True, False]}),
+ '{"A": [1, 2, 3, true, false]}'
+ )
+
+ def test_result_deserialize_default(self):
+ from sqlalchemy.engine import default
+
+ dialect = default.DefaultDialect()
+ proc = self.test_table.c.test_column.type._cached_result_processor(
+ dialect, None)
+ eq_(
+ proc('{"A": [1, 2, 3, true, false]}'),
+ {"A": [1, 2, 3, True, False]}
+ )
+
+ # This test is a bit misleading -- in real life you will need to cast to do anything
+ def test_where_getitem(self):
+ self._test_where(
+ self.jsoncol['bar'] == None,
+ "(test_table.test_column -> %(test_column_1)s) IS NULL"
+ )
+
+ def test_where_path(self):
+ self._test_where(
+ self.jsoncol.get_path('{"foo", 1}') == None,
+ "(test_table.test_column #> %(test_column_1)s) IS NULL"
+ )
+
+ def test_where_getitem_as_text(self):
+ self._test_where(
+ self.jsoncol.get_item_as_text('bar') == None,
+ "(test_table.test_column ->> %(test_column_1)s) IS NULL"
+ )
+
+ def test_where_path_as_text(self):
+ self._test_where(
+ self.jsoncol.get_path_as_text('{"foo", 1}') == None,
+ "(test_table.test_column #>> %(test_column_1)s) IS NULL"
+ )
+
+ def test_cols_get(self):
+ self._test_cols(
+ self.jsoncol['foo'],
+ "test_table.test_column -> %(test_column_1)s AS anon_1",
+ True
+ )
+
+
+class JSONRoundTripTest(fixtures.TablesTest):
@zzzeek Owner
zzzeek added a note

great job on this

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
+ __only_on__ = 'postgresql'
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('data_table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('name', String(30), nullable=False),
+ Column('data', JSON)
+ )
+
+ def _fixture_data(self, engine):
+ data_table = self.tables.data_table
+ engine.execute(
+ data_table.insert(),
+ {'name': 'r1', 'data': {"k1": "r1v1", "k2": "r1v2"}},
+ {'name': 'r2', 'data': {"k1": "r2v1", "k2": "r2v2"}},
+ {'name': 'r3', 'data': {"k1": "r3v1", "k2": "r3v2"}},
+ {'name': 'r4', 'data': {"k1": "r4v1", "k2": "r4v2"}},
+ {'name': 'r5', 'data': {"k1": "r5v1", "k2": "r5v2"}},
+ )
+
+ def _assert_data(self, compare):
+ data = testing.db.execute(
+ select([self.tables.data_table.c.data]).
+ order_by(self.tables.data_table.c.name)
+ ).fetchall()
+ eq_([d for d, in data], compare)
+
+ def _test_insert(self, engine):
+ engine.execute(
+ self.tables.data_table.insert(),
+ {'name': 'r1', 'data': {"k1": "r1v1", "k2": "r1v2"}}
+ )
+ self._assert_data([{"k1": "r1v1", "k2": "r1v2"}])
+
+ def _non_native_engine(self):
+ if testing.against("postgresql+psycopg2"):
+ engine = engines.testing_engine()
+ else:
+ engine = testing.db
+ engine.connect()
+ return engine
+
+ def test_reflect(self):
+ from sqlalchemy import inspect
+ insp = inspect(testing.db)
+ cols = insp.get_columns('data_table')
+ assert isinstance(cols[2]['type'], JSON)
+
+ @testing.only_on("postgresql+psycopg2")
+ def test_insert_native(self):
+ engine = testing.db
+ self._test_insert(engine)
+
+ def test_insert_python(self):
+ engine = self._non_native_engine()
+ self._test_insert(engine)
+
+ @testing.only_on("postgresql+psycopg2")
+ def test_criterion_native(self):
+ engine = testing.db
+ self._fixture_data(engine)
+ self._test_criterion(engine)
+
+ def test_criterion_python(self):
+ engine = self._non_native_engine()
+ self._fixture_data(engine)
+ self._test_criterion(engine)
+
+ def test_path_query(self):
+ engine = testing.db
+ self._fixture_data(engine)
+ data_table = self.tables.data_table
+ result = engine.execute(
+ select([data_table.c.data]).where(
+ data_table.c.data.get_path_as_text('{k1}') == 'r3v1'
+ )
+ ).first()
+ eq_(result, ({'k1': 'r3v1', 'k2': 'r3v2'},))
+
+ def test_query_returned_as_text(self):
+ engine = testing.db
+ self._fixture_data(engine)
+ data_table = self.tables.data_table
+ result = engine.execute(
+ select([data_table.c.data.get_item_as_text('k1')])
+ ).first()
+ assert isinstance(result[0], basestring)
+
+ def _test_criterion(self, engine):
+ data_table = self.tables.data_table
+ result = engine.execute(
+ select([data_table.c.data]).where(
+ data_table.c.data.get_item_as_text('k1') == 'r3v1'
+ )
+ ).first()
+ eq_(result, ({'k1': 'r3v1', 'k2': 'r3v2'},))
Something went wrong with that request. Please try again.