Add type accessors for JSON indexed/pathed element access

Added new accessors to expressions of type :class:`.JSON` to allow for
specific datatype access and comparison, covering strings, integers,
numeric, boolean elements.   This revises the documented approach of
CASTing to string when comparing values, instead adding specific
functionality into the PostgreSQL, SQlite, MySQL dialects to reliably
deliver these basic types in all cases.

The change also delivers a new feature to the test exclusions
system so that combinations and exclusions can be used together.

Fixes: #4276
Change-Id: Ica5a926c060feb40a0a7cd60b9d6e061d7825728

7 files changed, 460 insertions, 144 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 05edb6310..fb123bc0f 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1207,17 +1207,63 @@ class MySQLCompiler(compiler.SQLCompiler):
     def visit_sysdate_func(self, fn, **kw):
         return "SYSDATE()"
 
-    def visit_json_getitem_op_binary(self, binary, operator, **kw):
-        return "JSON_EXTRACT(%s, %s)" % (
+    def _render_json_extract_from_binary(self, binary, operator, **kw):
+        # note we are intentionally calling upon the process() calls in the
+        # order in which they appear in the SQL String as this is used
+        # by positional parameter rendering
+
+        if binary.type._type_affinity is sqltypes.JSON:
+            return "JSON_EXTRACT(%s, %s)" % (
+                self.process(binary.left, **kw),
+                self.process(binary.right, **kw),
+            )
+
+        # for non-JSON, MySQL doesn't handle JSON null at all so it has to
+        # be explicit
+        case_expression = "CASE JSON_EXTRACT(%s, %s) WHEN 'null' THEN NULL" % (
             self.process(binary.left, **kw),
             self.process(binary.right, **kw),
         )
 
+        if binary.type._type_affinity is sqltypes.Integer:
+            type_expression = (
+                "ELSE CAST(JSON_EXTRACT(%s, %s) AS SIGNED INTEGER)"
+                % (
+                    self.process(binary.left, **kw),
+                    self.process(binary.right, **kw),
+                )
+            )
+        elif binary.type._type_affinity is sqltypes.Numeric:
+            # FLOAT / REAL not added in MySQL til 8.0.17
+            type_expression = (
+                "ELSE CAST(JSON_EXTRACT(%s, %s) AS DECIMAL(10, 6))"
+                % (
+                    self.process(binary.left, **kw),
+                    self.process(binary.right, **kw),
+                )
+            )
+        elif binary.type._type_affinity is sqltypes.Boolean:
+            # the NULL handling is particularly weird with boolean, so
+            # explicitly return true/false constants
+            type_expression = "WHEN true THEN true ELSE false"
+        elif binary.type._type_affinity is sqltypes.String:
+            # this fails with a JSON value that's a four byte unicode
+            # string.  SQLite has the same problem at the moment
+            type_expression = "ELSE JSON_UNQUOTE(JSON_EXTRACT(%s, %s))" % (
+                self.process(binary.left, **kw),
+                self.process(binary.right, **kw),
+            )
+        else:
+            # other affinity....this is not expected right now
+            type_expression = "ELSE JSON_EXTRACT(%s, %s)"
+
+        return case_expression + " " + type_expression + " END"
+
+    def visit_json_getitem_op_binary(self, binary, operator, **kw):
+        return self._render_json_extract_from_binary(binary, operator, **kw)
+
     def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
-        return "JSON_EXTRACT(%s, %s)" % (
-            self.process(binary.left, **kw),
-            self.process(binary.right, **kw),
-        )
+        return self._render_json_extract_from_binary(binary, operator, **kw)
 
     def visit_on_duplicate_key_update(self, on_duplicate, **kw):
         if on_duplicate._parameter_ordering:
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index d6fd2623b..6d97033d0 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1559,13 +1559,36 @@ class PGCompiler(compiler.SQLCompiler):
             self.process(element.stop, **kw),
         )
 
-    def visit_json_getitem_op_binary(self, binary, operator, **kw):
+    def visit_json_getitem_op_binary(
+        self, binary, operator, _cast_applied=False, **kw
+    ):
+        if (
+            not _cast_applied
+            and binary.type._type_affinity is not sqltypes.JSON
+        ):
+            kw["_cast_applied"] = True
+            return self.process(sql.cast(binary, binary.type), **kw)
+
         kw["eager_grouping"] = True
-        return self._generate_generic_binary(binary, " -> ", **kw)
 
-    def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
+        return self._generate_generic_binary(
+            binary, " -> " if not _cast_applied else " ->> ", **kw
+        )
+
+    def visit_json_path_getitem_op_binary(
+        self, binary, operator, _cast_applied=False, **kw
+    ):
+        if (
+            not _cast_applied
+            and binary.type._type_affinity is not sqltypes.JSON
+        ):
+            kw["_cast_applied"] = True
+            return self.process(sql.cast(binary, binary.type), **kw)
+
         kw["eager_grouping"] = True
-        return self._generate_generic_binary(binary, " #> ", **kw)
+        return self._generate_generic_binary(
+            binary, " #> " if not _cast_applied else " #>> ", **kw
+        )
 
     def visit_getitem_binary(self, binary, operator, **kw):
         return "%s[%s]" % (
diff --git a/lib/sqlalchemy/dialects/postgresql/json.py b/lib/sqlalchemy/dialects/postgresql/json.py
index b8e3ac23d..ef48f3d86 100644
--- a/lib/sqlalchemy/dialects/postgresql/json.py
+++ b/lib/sqlalchemy/dialects/postgresql/json.py
@@ -123,11 +123,17 @@ class JSON(sqltypes.JSON):
 
         data_table.c.data['some key'].astext == 'some value'
 
+      Note that equivalent functionality is available via the
+      :attr:`.JSON.Comparator.as_string` accessor.
+
     * Index operations with CAST
       (equivalent to ``CAST(col ->> ['some key'] AS <type>)``)::
 
         data_table.c.data['some key'].astext.cast(Integer) == 5
 
+      Note that equivalent functionality is available via the
+      :attr:`.JSON.Comparator.as_integer` and similar accessors.
+
     * Path index operations (the ``#>`` operator)::
 
         data_table.c.data[('key_1', 'key_2', 5, ..., 'key_n')]
@@ -214,7 +220,6 @@ class JSON(sqltypes.JSON):
                 :meth:`.ColumnElement.cast`
 
             """
-
             if isinstance(self.expr.right.type, sqltypes.JSON.JSONPathType):
                 return self.expr.left.operate(
                     JSONPATH_ASTEXT,
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 02d44a260..2685a9243 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1012,13 +1012,23 @@ class SQLiteCompiler(compiler.SQLCompiler):
         )
 
     def visit_json_getitem_op_binary(self, binary, operator, **kw):
-        return "JSON_QUOTE(JSON_EXTRACT(%s, %s))" % (
+        if binary.type._type_affinity is sqltypes.JSON:
+            expr = "JSON_QUOTE(JSON_EXTRACT(%s, %s))"
+        else:
+            expr = "JSON_EXTRACT(%s, %s)"
+
+        return expr % (
             self.process(binary.left, **kw),
             self.process(binary.right, **kw),
         )
 
     def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
-        return "JSON_QUOTE(JSON_EXTRACT(%s, %s))" % (
+        if binary.type._type_affinity is sqltypes.JSON:
+            expr = "JSON_QUOTE(JSON_EXTRACT(%s, %s))"
+        else:
+            expr = "JSON_EXTRACT(%s, %s)"
+
+        return expr % (
             self.process(binary.left, **kw),
             self.process(binary.right, **kw),
         )
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 7829eb4d0..751b0d0ae 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -1961,7 +1961,9 @@ class JSON(Indexable, TypeEngine):
                 data = {"key1": "value1", "key2": "value2"}
             )
 
-    The base :class:`.types.JSON` provides these operations:
+    **JSON-Specific Expression Operators**
+
+    The :class:`.types.JSON` datatype provides these additional SQL operations:
 
     * Keyed index operations::
 
@@ -1975,63 +1977,70 @@ class JSON(Indexable, TypeEngine):
 
         data_table.c.data[('key_1', 'key_2', 5, ..., 'key_n')]
 
-    Additional operations are available from the dialect-specific versions
+    * Data casters for specific JSON element types, subsequent to an index
+      or path operation being invoked::
+
+        data_table.c.data["some key"].as_integer()
+
+      .. versionadded:: 1.3.11
+
+    Additional operations may be available from the dialect-specific versions
     of :class:`.types.JSON`, such as :class:`.postgresql.JSON` and
-    :class:`.postgresql.JSONB`, each of which offer more operators than
-    just the basic type.
+    :class:`.postgresql.JSONB` which both offer additional PostgreSQL-specific
+    operations.
+
+    **Casting JSON Elements to Other Types**
+
+    Index operations, i.e. those invoked by calling upon the expression using
+    the Python bracket operator as in ``some_column['some key']``, return an
+    expression object whose type defaults to :class:`.JSON` by default, so that
+    further JSON-oriented instructions may be called upon the result type.
+    However, it is likely more common that an index operation is expected
+    to return a specific scalar element, such as a string or integer.  In
+    order to provide access to these elements in a backend-agnostic way,
+    a series of data casters are provided:
 
-    Index operations return an expression object whose type defaults to
-    :class:`.JSON` by default, so that further JSON-oriented instructions may
-    be called upon the result type.   Note that there are backend-specific
-    idiosyncrasies here, including that the PostgreSQL database does not
-    generally compare a "json" to a "json" structure without type casts.  These
-    idiosyncrasies can be accommodated in a backend-neutral way by making
-    explicit use of the :func:`.cast` and :func:`.type_coerce` constructs.
-    Comparison of specific index elements of a :class:`.JSON` object to other
-    objects works best if the **left hand side is CAST to a string** and the
-    **right hand side is rendered as a JSON string**; a future SQLAlchemy
-    feature such as a generic "astext" modifier may simplify this at some
-    point:
+    * :meth:`.JSON.Comparator.as_string` - return the element as a string
 
-    * **Compare an element of a JSON structure to a string**::
+    * :meth:`.JSON.Comparator.as_boolean` - return the element as a boolean
 
-        from sqlalchemy import cast, type_coerce
-        from sqlalchemy import String, JSON
+    * :meth:`.JSON.Comparator.as_float` - return the element as a float
 
-        cast(
-            data_table.c.data['some_key'], String
-        ) == '"some_value"'
+    * :meth:`.JSON.Comparator.as_integer` - return the element as an integer
 
-        cast(
-            data_table.c.data['some_key'], String
-        ) == type_coerce("some_value", JSON)
+    These data casters are implemented by supporting dialects in order to
+    assure that comparisons to the above types will work as expected, such as::
 
-    * **Compare an element of a JSON structure to an integer**::
+        # integer comparison
+        data_table.c.data["some_integer_key"].as_integer() == 5
 
-        from sqlalchemy import cast, type_coerce
-        from sqlalchemy import String, JSON
+        # boolean comparison
+        data_table.c.data["some_boolean"].as_boolean() == True
 
-        cast(data_table.c.data['some_key'], String) == '55'
+    .. versionadded:: 1.3.11 Added type-specific casters for the basic JSON
+       data element types.
 
-        cast(
-            data_table.c.data['some_key'], String
-        ) == type_coerce(55, JSON)
+    .. note::
+
+        The data caster functions are new in version 1.3.11, and supersede
+        the previous documented approaches of using CAST; for reference,
+        this looked like::
+
+           from sqlalchemy import cast, type_coerce
+           from sqlalchemy import String, JSON
+           cast(
+               data_table.c.data['some_key'], String
+           ) == type_coerce(55, JSON)
 
-    * **Compare an element of a JSON structure to some other JSON structure**
-      - note that Python dictionaries are typically not ordered so care should
-      be taken here to assert that the JSON structures are identical::
+        The above case now works directly as::
 
-        from sqlalchemy import cast, type_coerce
-        from sqlalchemy import String, JSON
-        import json
+            data_table.c.data['some_key'].as_integer() == 5
 
-        cast(
-            data_table.c.data['some_key'], String
-        ) == json.dumps({"foo": "bar"})
+        For details on the previous comparison approach within the 1.3.x
+        series, see the documentation for SQLAlchemy 1.2 or the included HTML
+        files in the doc/ directory of the version's distribution.
 
-        cast(
-            data_table.c.data['some_key'], String
-        ) == type_coerce({"foo": "bar"}, JSON)
+    **Detecting Changes in JSON columns when using the ORM**
 
     The :class:`.JSON` type, when used with the SQLAlchemy ORM, does not
     detect in-place mutations to the structure.  In order to detect these, the
@@ -2040,6 +2049,8 @@ class JSON(Indexable, TypeEngine):
     will be detected by the unit of work.  See the example at :class:`.HSTORE`
     for a simple example involving a dictionary.
 
+    **Support for JSON null vs. SQL NULL**
+
     When working with NULL values, the :class:`.JSON` type recommends the
     use of two specific constants in order to differentiate between a column
     that evaluates to SQL NULL, e.g. no value, vs. the JSON-encoded string
@@ -2064,6 +2075,8 @@ class JSON(Indexable, TypeEngine):
     values, but care must be taken as to the value of the
     :paramref:`.JSON.none_as_null` in these cases.
 
+    **Customizing the JSON Serializer**
+
     The JSON serializer and deserializer used by :class:`.JSON` defaults to
     Python's ``json.dumps`` and ``json.loads`` functions; in the case of the
     psycopg2 dialect, psycopg2 may be using its own custom loader function.
@@ -2092,6 +2105,8 @@ class JSON(Indexable, TypeEngine):
 
         :class:`.mysql.JSON`
 
+        :class:`.sqlite.JSON`
+
     .. versionadded:: 1.1
 
 
@@ -2249,6 +2264,101 @@ class JSON(Indexable, TypeEngine):
 
             return operator, index, self.type
 
+        def as_boolean(self):
+            """Cast an indexed value as boolean.
+
+            e.g.::
+
+                stmt = select([
+                    mytable.c.json_column['some_data'].as_boolean()
+                ]).where(
+                    mytable.c.json_column['some_data'].as_boolean() == True
+                )
+
+            .. versionadded:: 1.3.11
+
+            """
+            return self._binary_w_type(Boolean(), "as_boolean")
+
+        def as_string(self):
+            """Cast an indexed value as string.
+
+            e.g.::
+
+                stmt = select([
+                    mytable.c.json_column['some_data'].as_string()
+                ]).where(
+                    mytable.c.json_column['some_data'].as_string() ==
+                    'some string'
+                )
+
+            .. versionadded:: 1.3.11
+
+            """
+            return self._binary_w_type(String(), "as_string")
+
+        def as_integer(self):
+            """Cast an indexed value as integer.
+
+            e.g.::
+
+                stmt = select([
+                    mytable.c.json_column['some_data'].as_integer()
+                ]).where(
+                    mytable.c.json_column['some_data'].as_integer() == 5
+                )
+
+            .. versionadded:: 1.3.11
+
+            """
+            return self._binary_w_type(Integer(), "as_integer")
+
+        def as_float(self):
+            """Cast an indexed value as float.
+
+            e.g.::
+
+                stmt = select([
+                    mytable.c.json_column['some_data'].as_float()
+                ]).where(
+                    mytable.c.json_column['some_data'].as_float() == 29.75
+                )
+
+            .. versionadded:: 1.3.11
+
+            """
+            # note there's no Numeric or Decimal support here yet
+            return self._binary_w_type(Float(), "as_float")
+
+        def as_json(self):
+            """Cast an indexed value as JSON.
+
+            This is the default behavior of indexed elements in any case.
+
+            Note that comparison of full JSON structures may not be
+            supported by all backends.
+
+            .. versionadded:: 1.3.11
+
+            """
+            return self.expr
+
+        def _binary_w_type(self, typ, method_name):
+            if not isinstance(
+                self.expr, elements.BinaryExpression
+            ) or self.expr.operator not in (
+                operators.json_getitem_op,
+                operators.json_path_getitem_op,
+            ):
+                raise exc.InvalidRequestError(
+                    "The JSON cast operator JSON.%s() only works with a JSON "
+                    "index expression e.g. col['q'].%s()"
+                    % (method_name, method_name)
+                )
+            expr = self.expr._clone()
+            expr.type = typ
+            return expr
+
     comparator_factory = Comparator
 
     @property
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index fd8d82690..5b26ac72e 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -686,6 +686,10 @@ class SuiteRequirements(Requirements):
         return self.json_type
 
     @property
+    def json_index_supplementary_unicode_element(self):
+        return exclusions.open()
+
+    @property
     def precision_numerics_general(self):
         """target backend has general support for moderately high-precision
         numerics."""
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index 435ab4689..bf5b18d0e 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -9,6 +9,7 @@ from .. import engines
 from .. import fixtures
 from .. import mock
 from ..assertions import eq_
+from ..assertions import is_
 from ..config import requirements
 from ..schema import Column
 from ..schema import Table
@@ -710,31 +711,6 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
 
     datatype = JSON
 
-    data1 = {"key1": "value1", "key2": "value2"}
-
-    data2 = {
-        "Key 'One'": "value1",
-        "key two": "value2",
-        "key three": "value ' three '",
-    }
-
-    data3 = {
-        "key1": [1, 2, 3],
-        "key2": ["one", "two", "three"],
-        "key3": [{"four": "five"}, {"six": "seven"}],
-    }
-
-    data4 = ["one", "two", "three"]
-
-    data5 = {
-        "nested": {
-            "elem1": [{"a": "b", "c": "d"}, {"e": "f", "g": "h"}],
-            "elem2": {"elem3": {"elem4": "elem5"}},
-        }
-    }
-
-    data6 = {"a": 5, "b": "some value", "c": {"foo": "bar"}}
-
     @classmethod
     def define_tables(cls, metadata):
         Table(
@@ -747,7 +723,7 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
         )
 
     def test_round_trip_data1(self):
-        self._test_round_trip(self.data1)
+        self._test_round_trip({"key1": "value1", "key2": "value2"})
 
     def _test_round_trip(self, data_element):
         data_table = self.tables.data_table
@@ -760,6 +736,97 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
 
         eq_(row, (data_element,))
 
+    def _index_fixtures(fn):
+        fn = testing.combinations(
+            ("boolean", True),
+            ("boolean", False),
+            ("boolean", None),
+            ("string", "some string"),
+            ("string", None),
+            ("string", util.u("réve illé")),
+            (
+                "string",
+                util.u("réve🐍 illé"),
+                testing.requires.json_index_supplementary_unicode_element,
+            ),
+            ("integer", 15),
+            ("integer", 1),
+            ("integer", 0),
+            ("integer", None),
+            ("float", 28.5),
+            ("float", None),
+            # TODO: how to test for comaprison
+            #        ("json", {"foo": "bar"}),
+            id_="sa",
+        )(fn)
+        return fn
+
+    @_index_fixtures
+    def test_index_typed_access(self, datatype, value):
+        data_table = self.tables.data_table
+        data_element = {"key1": value}
+        with config.db.connect() as conn:
+            conn.execute(
+                data_table.insert(),
+                {
+                    "name": "row1",
+                    "data": data_element,
+                    "nulldata": data_element,
+                },
+            )
+
+            expr = data_table.c.data["key1"]
+            expr = getattr(expr, "as_%s" % datatype)()
+
+            roundtrip = conn.scalar(select([expr]))
+            eq_(roundtrip, value)
+            if util.py3k:  # skip py2k to avoid comparing unicode to str etc.
+                is_(type(roundtrip), type(value))
+
+    @_index_fixtures
+    def test_index_typed_comparison(self, datatype, value):
+        data_table = self.tables.data_table
+        data_element = {"key1": value}
+        with config.db.connect() as conn:
+            conn.execute(
+                data_table.insert(),
+                {
+                    "name": "row1",
+                    "data": data_element,
+                    "nulldata": data_element,
+                },
+            )
+
+            expr = data_table.c.data["key1"]
+            expr = getattr(expr, "as_%s" % datatype)()
+
+            row = conn.execute(select([expr]).where(expr == value)).first()
+
+            # make sure we get a row even if value is None
+            eq_(row, (value,))
+
+    @_index_fixtures
+    def test_path_typed_comparison(self, datatype, value):
+        data_table = self.tables.data_table
+        data_element = {"key1": {"subkey1": value}}
+        with config.db.connect() as conn:
+            conn.execute(
+                data_table.insert(),
+                {
+                    "name": "row1",
+                    "data": data_element,
+                    "nulldata": data_element,
+                },
+            )
+
+            expr = data_table.c.data[("key1", "subkey1")]
+            expr = getattr(expr, "as_%s" % datatype)()
+
+            row = conn.execute(select([expr]).where(expr == value)).first()
+
+            # make sure we get a row even if value is None
+            eq_(row, (value,))
+
     def test_round_trip_custom_json(self):
         data_table = self.tables.data_table
         data_element = {"key1": "data1"}
@@ -840,6 +907,112 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
 
             eq_(conn.scalar(select([col])), None)
 
+    def test_unicode_round_trip(self):
+        # note we include Unicode supplementary characters as well
+        with config.db.connect() as conn:
+            conn.execute(
+                self.tables.data_table.insert(),
+                {
+                    "name": "r1",
+                    "data": {
+                        util.u("réve🐍 illé"): util.u("réve🐍 illé"),
+                        "data": {"k1": util.u("drôl🐍e")},
+                    },
+                },
+            )
+
+            eq_(
+                conn.scalar(select([self.tables.data_table.c.data])),
+                {
+                    util.u("réve🐍 illé"): util.u("réve🐍 illé"),
+                    "data": {"k1": util.u("drôl🐍e")},
+                },
+            )
+
+    def test_eval_none_flag_orm(self):
+
+        Base = declarative_base()
+
+        class Data(Base):
+            __table__ = self.tables.data_table
+
+        s = Session(testing.db)
+
+        d1 = Data(name="d1", data=None, nulldata=None)
+        s.add(d1)
+        s.commit()
+
+        s.bulk_insert_mappings(
+            Data, [{"name": "d2", "data": None, "nulldata": None}]
+        )
+        eq_(
+            s.query(
+                cast(self.tables.data_table.c.data, String()),
+                cast(self.tables.data_table.c.nulldata, String),
+            )
+            .filter(self.tables.data_table.c.name == "d1")
+            .first(),
+            ("null", None),
+        )
+        eq_(
+            s.query(
+                cast(self.tables.data_table.c.data, String()),
+                cast(self.tables.data_table.c.nulldata, String),
+            )
+            .filter(self.tables.data_table.c.name == "d2")
+            .first(),
+            ("null", None),
+        )
+
+
+class JSONStringCastIndexTest(_LiteralRoundTripFixture, fixtures.TablesTest):
+    """test JSON index access with "cast to string", which we have documented
+    for a long time as how to compare JSON values, but is ultimately not
+    reliable in all cases.
+
+    """
+
+    __requires__ = ("json_type",)
+    __backend__ = True
+
+    datatype = JSON
+
+    data1 = {"key1": "value1", "key2": "value2"}
+
+    data2 = {
+        "Key 'One'": "value1",
+        "key two": "value2",
+        "key three": "value ' three '",
+    }
+
+    data3 = {
+        "key1": [1, 2, 3],
+        "key2": ["one", "two", "three"],
+        "key3": [{"four": "five"}, {"six": "seven"}],
+    }
+
+    data4 = ["one", "two", "three"]
+
+    data5 = {
+        "nested": {
+            "elem1": [{"a": "b", "c": "d"}, {"e": "f", "g": "h"}],
+            "elem2": {"elem3": {"elem4": "elem5"}},
+        }
+    }
+
+    data6 = {"a": 5, "b": "some value", "c": {"foo": "bar"}}
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "data_table",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("name", String(30), nullable=False),
+            Column("data", cls.datatype),
+            Column("nulldata", cls.datatype(none_as_null=True)),
+        )
+
     def _criteria_fixture(self):
         config.db.execute(
             self.tables.data_table.insert(),
@@ -869,7 +1042,7 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
 
                 eq_(conn.scalar(literal_sql), expected)
 
-    def test_crit_spaces_in_key(self):
+    def test_string_cast_crit_spaces_in_key(self):
         name = self.tables.data_table.c.name
         col = self.tables.data_table.c["data"]
 
@@ -885,7 +1058,7 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
         )
 
     @config.requirements.json_array_indexes
-    def test_crit_simple_int(self):
+    def test_string_cast_crit_simple_int(self):
         name = self.tables.data_table.c.name
         col = self.tables.data_table.c["data"]
 
@@ -896,13 +1069,13 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
             and_(name == "r4", cast(col[1], String) == '"two"'), "r4"
         )
 
-    def test_crit_mixed_path(self):
+    def test_string_cast_crit_mixed_path(self):
         col = self.tables.data_table.c["data"]
         self._test_index_criteria(
             cast(col[("key3", 1, "six")], String) == '"seven"', "r3"
         )
 
-    def test_crit_string_path(self):
+    def test_string_cast_crit_string_path(self):
         col = self.tables.data_table.c["data"]
         self._test_index_criteria(
             cast(col[("nested", "elem2", "elem3", "elem4")], String)
@@ -910,7 +1083,7 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
             "r5",
         )
 
-    def test_crit_against_string_basic(self):
+    def test_string_cast_crit_against_string_basic(self):
         name = self.tables.data_table.c.name
         col = self.tables.data_table.c["data"]
 
@@ -949,67 +1122,12 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest):
             test_literal=False,
         )
 
-    def test_unicode_round_trip(self):
-        with config.db.connect() as conn:
-            conn.execute(
-                self.tables.data_table.insert(),
-                {
-                    "name": "r1",
-                    "data": {
-                        util.u("réve🐍 illé"): util.u("réve🐍 illé"),
-                        "data": {"k1": util.u("drôl🐍e")},
-                    },
-                },
-            )
-
-            eq_(
-                conn.scalar(select([self.tables.data_table.c.data])),
-                {
-                    util.u("réve🐍 illé"): util.u("réve🐍 illé"),
-                    "data": {"k1": util.u("drôl🐍e")},
-                },
-            )
-
-    def test_eval_none_flag_orm(self):
-
-        Base = declarative_base()
-
-        class Data(Base):
-            __table__ = self.tables.data_table
-
-        s = Session(testing.db)
-
-        d1 = Data(name="d1", data=None, nulldata=None)
-        s.add(d1)
-        s.commit()
-
-        s.bulk_insert_mappings(
-            Data, [{"name": "d2", "data": None, "nulldata": None}]
-        )
-        eq_(
-            s.query(
-                cast(self.tables.data_table.c.data, String()),
-                cast(self.tables.data_table.c.nulldata, String),
-            )
-            .filter(self.tables.data_table.c.name == "d1")
-            .first(),
-            ("null", None),
-        )
-        eq_(
-            s.query(
-                cast(self.tables.data_table.c.data, String()),
-                cast(self.tables.data_table.c.nulldata, String),
-            )
-            .filter(self.tables.data_table.c.name == "d2")
-            .first(),
-            ("null", None),
-        )
-
 
 __all__ = (
     "UnicodeVarcharTest",
     "UnicodeTextTest",
     "JSONTest",
+    "JSONStringCastIndexTest",
     "DateTest",
     "DateTimeTest",
     "TextTest",