--- /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
