--- /dev/null
+.. change::
+ :tags: usecase, sql
+ :tickets: 4276
+
+ 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.
+
+ .. seealso::
+
+ :class:`.JSON`
+
+ :meth:`.JSON.Comparator.as_string`
+
+ :meth:`.JSON.Comparator.as_boolean`
+
+ :meth:`.JSON.Comparator.as_float`
+
+ :meth:`.JSON.Comparator.as_integer`
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:
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]" % (
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')]
:meth:`.ColumnElement.cast`
"""
-
if isinstance(self.expr.right.type, sqltypes.JSON.JSONPathType):
return self.expr.left.operate(
JSONPATH_ASTEXT,
)
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),
)
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::
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
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
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.
:class:`.mysql.JSON`
+ :class:`.sqlite.JSON`
+
.. versionadded:: 1.1
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
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
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
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(
)
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
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"}
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(),
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"]
)
@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"]
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)
"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"]
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",
from sqlalchemy.testing.assertions import ComparesTables
from sqlalchemy.testing.assertions import eq_
from sqlalchemy.testing.assertions import is_
+from sqlalchemy.testing.suite import test_types as suite
from sqlalchemy.testing.util import round_decimal
@testing.requires.postgresql_utf8_server_encoding
def test_unicode_round_trip_native(self):
super(JSONBRoundTripTest, self).test_unicode_round_trip_native()
+
+
+class JSONBSuiteTest(suite.JSONTest):
+ __requires__ = ("postgresql_jsonb",)
+
+ datatype = JSONB
+
+
+class JSONBCastSuiteTest(suite.JSONStringCastIndexTest):
+ __requires__ = ("postgresql_jsonb",)
+
+ datatype = JSONB
]
)
+ @property
+ def json_index_supplementary_unicode_element(self):
+ # for sqlite see https://bugs.python.org/issue38749
+ return skip_if(
+ [
+ lambda config: against(config, "mysql")
+ and config.db.dialect._is_mariadb,
+ "sqlite",
+ ]
+ )
+
def _sqlite_json(self, config):
if not against(config, "sqlite >= 3.9"):
return False
from sqlalchemy import LargeBinary
from sqlalchemy import literal_column
from sqlalchemy import not_
+from sqlalchemy import Numeric
from sqlalchemy import or_
from sqlalchemy import String
from sqlalchemy import testing
col = Column("x", MyOtherType())
self.assert_compile(col[5], "x $$> :x_1", checkparams={"x_1": 5})
+ def _caster_combinations(fn):
+ return testing.combinations(
+ ("integer", Integer),
+ ("boolean", Boolean),
+ ("float", Numeric),
+ ("string", String),
+ )(fn)
+
+ @_caster_combinations
+ def test_cast_ops(self, caster, expected_type):
+ expr = Column("x", JSON)["foo"]
+
+ expr = getattr(expr, "as_%s" % caster)()
+ is_(expr.type._type_affinity, expected_type)
+
+ @_caster_combinations
+ def test_cast_ops_unsupported_on_non_binary(self, caster, expected_type):
+ expr = Column("x", JSON)
+
+ meth = getattr(expr, "as_%s" % caster)
+
+ assert_raises_message(
+ exc.InvalidRequestError,
+ r"The JSON cast operator JSON.as_%s\(\) only works" % caster,
+ meth,
+ )
+
+ @_caster_combinations
+ def test_cast_ops_unsupported_on_non_json_binary(
+ self, caster, expected_type
+ ):
+ expr = Column("x", JSON) + {"foo": "bar"}
+
+ meth = getattr(expr, "as_%s" % caster)
+
+ assert_raises_message(
+ exc.InvalidRequestError,
+ r"The JSON cast operator JSON.as_%s\(\) only works" % caster,
+ meth,
+ )
+
class ArrayIndexOpTest(fixtures.TestBase, testing.AssertsCompiledSQL):
def setUp(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
