Fixed regression where the row objects returned for ORM queries, which are
now the normal :class:`_sql.Row` objects, would not be interpreted by the
:meth:`_sql.ColumnOperators.in_` operator as tuple values to be broken out
into individual bound parameters, and would instead pass them as single
values to the driver leading to failures. The change to the "expanding IN"
system now accommodates for the expression already being of type
:class:`.TupleType` and treats values accordingly if so. In the uncommon
case of using "tuple-in" with an untyped statement such as a textual
statement with no typing information, a tuple value is detected for values
that implement ``collections.abc.Sequence``, but that are not ``str`` or
``bytes``, as always when testing for ``Sequence``.
Added :class:`.TupleType` to the top level ``sqlalchemy`` import namespace.
Fixes: #7292
Change-Id: I8286387e3b3c3752b3bd4ae3560d4f31172acc22
(cherry picked from commit
0c44a1e77cfde0f841a4a64140314c6b833efdab)
--- /dev/null
+.. change::
+ :tags: bug, sql, regression
+ :tickets: 7292
+
+ Fixed regression where the row objects returned for ORM queries, which are
+ now the normal :class:`_sql.Row` objects, would not be interpreted by the
+ :meth:`_sql.ColumnOperators.in_` operator as tuple values to be broken out
+ into individual bound parameters, and would instead pass them as single
+ values to the driver leading to failures. The change to the "expanding IN"
+ system now accommodates for the expression already being of type
+ :class:`.TupleType` and treats values accordingly if so. In the uncommon
+ case of using "tuple-in" with an untyped statement such as a textual
+ statement with no typing information, a tuple value is detected for values
+ that implement ``collections.abc.Sequence``, but that are not ``str`` or
+ ``bytes``, as always when testing for ``Sequence``.
+
+.. change::
+ :tags: usecase, sql
+
+ Added :class:`.TupleType` to the top level ``sqlalchemy`` import namespace.
\ No newline at end of file
from .types import TIME
from .types import Time
from .types import TIMESTAMP
+from .types import TupleType
from .types import TypeDecorator
from .types import Unicode
from .types import UnicodeText
[parameter.type], parameter.expand_op
)
- elif isinstance(values[0], (tuple, list)):
- assert typ_dialect_impl._is_tuple_type
+ elif typ_dialect_impl._is_tuple_type or (
+ typ_dialect_impl._isnull
+ and isinstance(values[0], util.collections_abc.Sequence)
+ and not isinstance(
+ values[0], util.string_types + util.binary_types
+ )
+ ):
+
replacement_expression = (
"VALUES " if self.dialect.tuple_in_values else ""
) + ", ".join(
for i, tuple_element in enumerate(values)
)
else:
- assert not typ_dialect_impl._is_tuple_type
replacement_expression = ", ".join(
self.render_literal_value(value, parameter.type)
for value in values
[parameter.type], parameter.expand_op
)
- elif (
- isinstance(values[0], (tuple, list))
- and not typ_dialect_impl._is_array
+ elif typ_dialect_impl._is_tuple_type or (
+ typ_dialect_impl._isnull
+ and isinstance(values[0], util.collections_abc.Sequence)
+ and not isinstance(
+ values[0], util.string_types + util.binary_types
+ )
):
+ assert not typ_dialect_impl._is_array
to_update = [
("%s_%s_%s" % (name, i, j), value)
for i, tuple_element in enumerate(values, 1)
def __init__(self, *types):
self._fully_typed = NULLTYPE not in types
- self.types = types
+ self.types = [
+ item_type() if isinstance(item_type, type) else item_type
+ for item_type in types
+ ]
def _resolve_values_to_types(self, value):
if self._fully_typed:
from ... import text
from ... import true
from ... import tuple_
+from ... import TupleType
from ... import union
from ... import util
from ... import values
from ...exc import DatabaseError
from ...exc import ProgrammingError
+from ...util import collections_abc
class CollateTest(fixtures.TablesTest):
)
self._assert_result(stmt, [])
+ def test_typed_str_in(self):
+ """test related to #7292.
+
+ as a type is given to the bound param, there is no ambiguity
+ to the type of element.
+
+ """
+
+ stmt = text(
+ "select id FROM some_table WHERE z IN :q ORDER BY id"
+ ).bindparams(bindparam("q", type_=String, expanding=True))
+ self._assert_result(
+ stmt,
+ [(2,), (3,), (4,)],
+ params={"q": ["z2", "z3", "z4"]},
+ )
+
+ def test_untyped_str_in(self):
+ """test related to #7292.
+
+ for untyped expression, we look at the types of elements.
+ Test for Sequence to detect tuple in. but not strings or bytes!
+ as always....
+
+ """
+
+ stmt = text(
+ "select id FROM some_table WHERE z IN :q ORDER BY id"
+ ).bindparams(bindparam("q", expanding=True))
+ self._assert_result(
+ stmt,
+ [(2,), (3,), (4,)],
+ params={"q": ["z2", "z3", "z4"]},
+ )
+
@testing.requires.tuple_in
def test_bound_in_two_tuple_bindparam(self):
table = self.tables.some_table
params={"q": [(2, "z2"), (3, "z3"), (4, "z4")]},
)
+ @testing.requires.tuple_in
+ def test_bound_in_heterogeneous_two_tuple_typed_bindparam_non_tuple(self):
+ class LikeATuple(collections_abc.Sequence):
+ def __init__(self, *data):
+ self._data = data
+
+ def __iter__(self):
+ return iter(self._data)
+
+ def __getitem__(self, idx):
+ return self._data[idx]
+
+ def __len__(self):
+ return len(self._data)
+
+ stmt = text(
+ "select id FROM some_table WHERE (x, z) IN :q ORDER BY id"
+ ).bindparams(
+ bindparam(
+ "q", type_=TupleType(Integer(), String()), expanding=True
+ )
+ )
+ self._assert_result(
+ stmt,
+ [(2,), (3,), (4,)],
+ params={
+ "q": [
+ LikeATuple(2, "z2"),
+ LikeATuple(3, "z3"),
+ LikeATuple(4, "z4"),
+ ]
+ },
+ )
+
+ @testing.requires.tuple_in
+ def test_bound_in_heterogeneous_two_tuple_text_bindparam_non_tuple(self):
+ # note this becomes ARRAY if we dont use expanding
+ # explicitly right now
+
+ class LikeATuple(collections_abc.Sequence):
+ def __init__(self, *data):
+ self._data = data
+
+ def __iter__(self):
+ return iter(self._data)
+
+ def __getitem__(self, idx):
+ return self._data[idx]
+
+ def __len__(self):
+ return len(self._data)
+
+ stmt = text(
+ "select id FROM some_table WHERE (x, z) IN :q ORDER BY id"
+ ).bindparams(bindparam("q", expanding=True))
+ self._assert_result(
+ stmt,
+ [(2,), (3,), (4,)],
+ params={
+ "q": [
+ LikeATuple(2, "z2"),
+ LikeATuple(3, "z3"),
+ LikeATuple(4, "z4"),
+ ]
+ },
+ )
+
def test_empty_set_against_integer_bindparam(self):
table = self.tables.some_table
stmt = (
"INTEGER",
"DATE",
"TIME",
+ "TupleType",
"String",
"Integer",
"SmallInteger",
from .sql.sqltypes import TIME
from .sql.sqltypes import Time
from .sql.sqltypes import TIMESTAMP
+from .sql.sqltypes import TupleType
from .sql.sqltypes import Unicode
from .sql.sqltypes import UnicodeText
from .sql.sqltypes import VARBINARY
from .compat import b64decode
from .compat import b64encode
from .compat import binary_type
+from .compat import binary_types
from .compat import byte_buffer
from .compat import callable
from .compat import cmp
from sqlalchemy.testing import ne_
from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.assertsql import CompiledSQL
+from sqlalchemy.types import ARRAY
from sqlalchemy.types import Boolean
from sqlalchemy.types import Integer
from sqlalchemy.types import String
def test_in_parameters_five(self):
def go(n1, n2):
stmt = lambdas.lambda_stmt(
- lambda: select(1).where(column("q").in_(n1))
+ lambda: select(1).where(column("q", ARRAY(String)).in_(n1))
)
- stmt += lambda s: s.where(column("y").in_(n2))
+ stmt += lambda s: s.where(column("y", ARRAY(String)).in_(n2))
return stmt
expr = go(["a", "b", "c"], ["d", "e", "f"])
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import true
+from sqlalchemy import tuple_
from sqlalchemy import type_coerce
from sqlalchemy import TypeDecorator
from sqlalchemy import util
connection.execute(users.insert(), r._mapping)
eq_(connection.execute(users.select()).fetchall(), [(1, "john")])
+ @testing.requires.tuple_in
+ def test_row_tuple_interpretation(self, connection):
+ """test #7292"""
+ users = self.tables.users
+
+ connection.execute(
+ users.insert(),
+ [
+ dict(user_id=1, user_name="u1"),
+ dict(user_id=2, user_name="u2"),
+ dict(user_id=3, user_name="u3"),
+ ],
+ )
+ rows = connection.execute(
+ select(users.c.user_id, users.c.user_name)
+ ).all()
+
+ # was previously needed
+ # rows = [(x, y) for x, y in rows]
+
+ new_stmt = (
+ select(users)
+ .where(tuple_(users.c.user_id, users.c.user_name).in_(rows))
+ .order_by(users.c.user_id)
+ )
+
+ eq_(
+ connection.execute(new_stmt).all(),
+ [(1, "u1"), (2, "u2"), (3, "u3")],
+ )
+
def test_result_as_args(self, connection):
users = self.tables.users
users2 = self.tables.users2
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
