__slots__ = ()
def _warn_for_scalar_subquery_coercion(self):
- util.warn_deprecated(
- "coercing SELECT object to scalar subquery in a "
- "column-expression context is deprecated in version 1.4; "
+ util.warn(
+ "implicitly coercing SELECT object to scalar subquery; "
"please use the .scalar_subquery() method to produce a scalar "
- "subquery. This automatic coercion will be removed in a "
- "future release.",
- version="1.4",
+ "subquery.",
)
def _implicit_coercions(
isinstance(resolved, selectable.Alias)
and resolved.element._is_select_statement
):
- return resolved.element
+ self._warn_for_implicit_coercion(resolved)
+ return self._post_coercion(resolved.element, **kw)
else:
- return resolved.select()
+ self._warn_for_implicit_coercion(resolved)
+ return self._post_coercion(resolved.select(), **kw)
else:
self._raise_for_expected(original_element, argname, resolved)
+ def _warn_for_implicit_coercion(self, elem):
+ util.warn(
+ "Coercing %s object into a select() for use in IN(); "
+ "please pass a select() construct explicitly"
+ % (elem.__class__.__name__)
+ )
+
def _literal_coercion(self, element, expr, operator, **kw):
if isinstance(element, collections_abc.Iterable) and not isinstance(
element, util.string_types
def _post_coercion(self, element, expr, operator, **kw):
if element._is_select_statement:
+ # for IN, we are doing scalar_subquery() coercion without
+ # a warning
return element.scalar_subquery()
elif isinstance(element, elements.ClauseList):
assert not len(element.clauses) == 0
from sqlalchemy import table
from sqlalchemy import testing
from sqlalchemy import text
-from sqlalchemy import update
from sqlalchemy import util
from sqlalchemy import VARCHAR
from sqlalchemy.engine import default
"ON myothertable.otherid = mytable.myid",
)
- def test_column_roles(self):
- stmt = select([self.table1.c.myid])
-
- for role in [
- roles.WhereHavingRole,
- roles.ExpressionElementRole,
- roles.ByOfRole,
- roles.OrderByRole,
- # roles.LabeledColumnExprRole
- ]:
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- coerced = coercions.expect(role, stmt)
- is_true(coerced.compare(stmt.scalar_subquery()))
-
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- coerced = coercions.expect(role, stmt.alias())
- is_true(coerced.compare(stmt.scalar_subquery()))
-
- def test_labeled_role(self):
- stmt = select([self.table1.c.myid])
-
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- coerced = coercions.expect(roles.LabeledColumnExprRole, stmt)
- is_true(coerced.compare(stmt.scalar_subquery().label(None)))
-
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- coerced = coercions.expect(
- roles.LabeledColumnExprRole, stmt.alias()
- )
- is_true(coerced.compare(stmt.scalar_subquery().label(None)))
-
- def test_scalar_select(self):
-
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- self.assert_compile(
- func.coalesce(select([self.table1.c.myid])),
- "coalesce((SELECT mytable.myid FROM mytable))",
- )
-
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- s = select([self.table1.c.myid]).alias()
- self.assert_compile(
- select([self.table1.c.myid]).where(self.table1.c.myid == s),
- "SELECT mytable.myid FROM mytable WHERE "
- "mytable.myid = (SELECT mytable.myid FROM "
- "mytable)",
- )
-
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- self.assert_compile(
- select([self.table1.c.myid]).where(s > self.table1.c.myid),
- "SELECT mytable.myid FROM mytable WHERE "
- "mytable.myid < (SELECT mytable.myid FROM "
- "mytable)",
- )
-
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- s = select([self.table1.c.myid]).alias()
- self.assert_compile(
- select([self.table1.c.myid]).where(self.table1.c.myid == s),
- "SELECT mytable.myid FROM mytable WHERE "
- "mytable.myid = (SELECT mytable.myid FROM "
- "mytable)",
- )
-
- with testing.expect_deprecated(
- "coercing SELECT object to scalar "
- "subquery in a column-expression context is deprecated"
- ):
- self.assert_compile(
- select([self.table1.c.myid]).where(s > self.table1.c.myid),
- "SELECT mytable.myid FROM mytable WHERE "
- "mytable.myid < (SELECT mytable.myid FROM "
- "mytable)",
- )
-
def test_standalone_alias(self):
with testing.expect_deprecated(
"Implicit coercion of SELECT and textual SELECT constructs"
self.assert_compile(
stmt, "UPDATE foo SET bar=%s LIMIT 10", dialect="mysql"
)
-
- @testing.fixture()
- def update_from_fixture(self):
- metadata = MetaData()
-
- mytable = Table(
- "mytable",
- metadata,
- Column("myid", Integer),
- Column("name", String(30)),
- Column("description", String(50)),
- )
- myothertable = Table(
- "myothertable",
- metadata,
- Column("otherid", Integer),
- Column("othername", String(30)),
- )
- return mytable, myothertable
-
- def test_correlated_update_two(self, update_from_fixture):
- table1, t2 = update_from_fixture
-
- mt = table1.alias()
- with testing.expect_deprecated(
- "coercing SELECT object to scalar subquery in a column-expression "
- "context is deprecated"
- ):
- u = update(
- table1,
- values={
- table1.c.name: select(
- [mt.c.name], mt.c.myid == table1.c.myid
- )
- },
- )
- self.assert_compile(
- u,
- "UPDATE mytable SET name=(SELECT mytable_1.name FROM "
- "mytable AS mytable_1 WHERE "
- "mytable_1.myid = mytable.myid)",
- )
-
- def test_correlated_update_three(self, update_from_fixture):
- table1, table2 = update_from_fixture
-
- # test against a regular constructed subquery
- s = select([table2], table2.c.otherid == table1.c.myid)
- with testing.expect_deprecated(
- "coercing SELECT object to scalar subquery in a column-expression "
- "context is deprecated"
- ):
- u = update(
- table1, table1.c.name == "jack", values={table1.c.name: s}
- )
- self.assert_compile(
- u,
- "UPDATE mytable SET name=(SELECT myothertable.otherid, "
- "myothertable.othername FROM myothertable WHERE "
- "myothertable.otherid = mytable.myid) "
- "WHERE mytable.name = :name_1",
- )
self._assert_types(expr.right._expanding_in_types)
+class InSelectableTest(fixtures.TestBase, testing.AssertsCompiledSQL):
+ __dialect__ = "default"
+
+ def test_in_select(self):
+ t = table("t", column("x"))
+
+ stmt = select([t.c.x])
+
+ self.assert_compile(column("q").in_(stmt), "q IN (SELECT t.x FROM t)")
+
+ def test_in_subquery_warning(self):
+ t = table("t", column("x"))
+
+ stmt = select([t.c.x]).subquery()
+
+ with expect_warnings(
+ r"Coercing Subquery object into a select\(\) for use in "
+ r"IN\(\); please pass a select\(\) construct explicitly",
+ ):
+ self.assert_compile(
+ column("q").in_(stmt),
+ "q IN (SELECT anon_1.x FROM "
+ "(SELECT t.x AS x FROM t) AS anon_1)",
+ )
+
+ def test_in_subquery_explicit(self):
+ t = table("t", column("x"))
+
+ stmt = select([t.c.x]).subquery()
+
+ self.assert_compile(
+ column("q").in_(stmt.select()),
+ "q IN (SELECT anon_1.x FROM "
+ "(SELECT t.x AS x FROM t) AS anon_1)",
+ )
+
+ def test_in_subquery_alias_implicit(self):
+ t = table("t", column("x"))
+
+ stmt = select([t.c.x]).subquery().alias()
+
+ with expect_warnings(
+ r"Coercing Alias object into a select\(\) for use in "
+ r"IN\(\); please pass a select\(\) construct explicitly",
+ ):
+ self.assert_compile(
+ column("q").in_(stmt),
+ "q IN (SELECT anon_1.x FROM (SELECT t.x AS x FROM t) "
+ "AS anon_1)",
+ )
+
+ def test_in_subquery_alias_explicit(self):
+ t = table("t", column("x"))
+
+ stmt = select([t.c.x]).subquery().alias()
+
+ self.assert_compile(
+ column("q").in_(stmt.select().scalar_subquery()),
+ "q IN (SELECT anon_1.x FROM (SELECT t.x AS x FROM t) AS anon_1)",
+ )
+
+ def test_in_table(self):
+ t = table("t", column("x"))
+
+ with expect_warnings(
+ r"Coercing TableClause object into a select\(\) for use in "
+ r"IN\(\); please pass a select\(\) construct explicitly",
+ ):
+ self.assert_compile(column("q").in_(t), "q IN (SELECT t.x FROM t)")
+
+ def test_in_table_alias(self):
+ t = table("t", column("x"))
+
+ with expect_warnings(
+ r"Coercing Alias object into a select\(\) for use in "
+ r"IN\(\); please pass a select\(\) construct explicitly",
+ ):
+ self.assert_compile(
+ column("q").in_(t.alias()), "q IN (SELECT t_1.x FROM t AS t_1)"
+ )
+
+ def test_in_cte_implicit(self):
+ t = table("t", column("x"))
+
+ stmt = select([t.c.x]).cte()
+
+ with expect_warnings(
+ r"Coercing CTE object into a select\(\) for use in "
+ r"IN\(\); please pass a select\(\) construct explicitly",
+ ):
+ s2 = select([column("q").in_(stmt)])
+
+ self.assert_compile(
+ s2,
+ "WITH anon_2 AS (SELECT t.x AS x FROM t) "
+ "SELECT q IN (SELECT anon_2.x FROM anon_2) AS anon_1",
+ )
+
+ def test_in_cte_explicit(self):
+ t = table("t", column("x"))
+
+ stmt = select([t.c.x]).cte()
+
+ s2 = select([column("q").in_(stmt.select().scalar_subquery())])
+
+ self.assert_compile(
+ s2,
+ "WITH anon_2 AS (SELECT t.x AS x FROM t) "
+ "SELECT q IN (SELECT anon_2.x FROM anon_2) AS anon_1",
+ )
+
+ def test_in_cte_select(self):
+ t = table("t", column("x"))
+
+ stmt = select([t.c.x]).cte()
+
+ s2 = select([column("q").in_(stmt.select())])
+
+ self.assert_compile(
+ s2,
+ "WITH anon_2 AS (SELECT t.x AS x FROM t) "
+ "SELECT q IN (SELECT anon_2.x FROM anon_2) AS anon_1",
+ )
+
+
class AnyAllTest(fixtures.TestBase, testing.AssertsCompiledSQL):
__dialect__ = "default"
from sqlalchemy import bindparam
from sqlalchemy import Column
+from sqlalchemy import column
from sqlalchemy import exc
+from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import select
+from sqlalchemy import String
from sqlalchemy import Table
+from sqlalchemy import table
from sqlalchemy import testing
from sqlalchemy import text
+from sqlalchemy import update
from sqlalchemy.schema import DDL
from sqlalchemy.schema import Sequence
from sqlalchemy.sql import ClauseElement
from sqlalchemy.sql import coercions
-from sqlalchemy.sql import column
from sqlalchemy.sql import false
from sqlalchemy.sql import False_
from sqlalchemy.sql import literal
from sqlalchemy.sql.selectable import SelectStatementGrouping
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_instance_of
)
def test_scalar_select_no_coercion(self):
- # this is also tested in test/sql/test_deprecations.py; when the
- # deprecation is turned to an error, those tests go away, and these
- # will assert the correct exception plus informative error message.
- assert_raises_message(
- exc.SADeprecationWarning,
- "coercing SELECT object to scalar subquery in a column-expression "
- "context is deprecated",
- expect,
- roles.LabeledColumnExprRole,
- select([column("q")]),
- )
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ expect(
+ roles.LabeledColumnExprRole, select([column("q")]),
+ )
- assert_raises_message(
- exc.SADeprecationWarning,
- "coercing SELECT object to scalar subquery in a column-expression "
- "context is deprecated",
- expect,
- roles.LabeledColumnExprRole,
- select([column("q")]).alias(),
- )
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ expect(
+ roles.LabeledColumnExprRole, select([column("q")]).alias(),
+ )
def test_statement_no_text_coercion(self):
assert_raises_message(
def test_columns_clause_role_neg(self):
self._test_role_neg_comparisons(roles.ColumnsClauseRole)
+
+
+class SubqueryCoercionsTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = "default"
+
+ table1 = table(
+ "mytable",
+ column("myid", Integer),
+ column("name", String),
+ column("description", String),
+ )
+
+ table2 = table(
+ "myothertable", column("otherid", Integer), column("othername", String)
+ )
+
+ def test_column_roles(self):
+ stmt = select([self.table1.c.myid])
+
+ for role in [
+ roles.WhereHavingRole,
+ roles.ExpressionElementRole,
+ roles.ByOfRole,
+ roles.OrderByRole,
+ # roles.LabeledColumnExprRole
+ ]:
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ coerced = coercions.expect(role, stmt)
+ is_true(coerced.compare(stmt.scalar_subquery()))
+
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ coerced = coercions.expect(role, stmt.alias())
+ is_true(coerced.compare(stmt.scalar_subquery()))
+
+ def test_labeled_role(self):
+ stmt = select([self.table1.c.myid])
+
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ coerced = coercions.expect(roles.LabeledColumnExprRole, stmt)
+ is_true(coerced.compare(stmt.scalar_subquery().label(None)))
+
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ coerced = coercions.expect(
+ roles.LabeledColumnExprRole, stmt.alias()
+ )
+ is_true(coerced.compare(stmt.scalar_subquery().label(None)))
+
+ def test_scalar_select(self):
+
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ self.assert_compile(
+ func.coalesce(select([self.table1.c.myid])),
+ "coalesce((SELECT mytable.myid FROM mytable))",
+ )
+
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ s = select([self.table1.c.myid]).alias()
+ self.assert_compile(
+ select([self.table1.c.myid]).where(self.table1.c.myid == s),
+ "SELECT mytable.myid FROM mytable WHERE "
+ "mytable.myid = (SELECT mytable.myid FROM "
+ "mytable)",
+ )
+
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ self.assert_compile(
+ select([self.table1.c.myid]).where(s > self.table1.c.myid),
+ "SELECT mytable.myid FROM mytable WHERE "
+ "mytable.myid < (SELECT mytable.myid FROM "
+ "mytable)",
+ )
+
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ s = select([self.table1.c.myid]).alias()
+ self.assert_compile(
+ select([self.table1.c.myid]).where(self.table1.c.myid == s),
+ "SELECT mytable.myid FROM mytable WHERE "
+ "mytable.myid = (SELECT mytable.myid FROM "
+ "mytable)",
+ )
+
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ self.assert_compile(
+ select([self.table1.c.myid]).where(s > self.table1.c.myid),
+ "SELECT mytable.myid FROM mytable WHERE "
+ "mytable.myid < (SELECT mytable.myid FROM "
+ "mytable)",
+ )
+
+ @testing.fixture()
+ def update_from_fixture(self):
+ metadata = MetaData()
+
+ mytable = Table(
+ "mytable",
+ metadata,
+ Column("myid", Integer),
+ Column("name", String(30)),
+ Column("description", String(50)),
+ )
+ myothertable = Table(
+ "myothertable",
+ metadata,
+ Column("otherid", Integer),
+ Column("othername", String(30)),
+ )
+ return mytable, myothertable
+
+ def test_correlated_update_two(self, update_from_fixture):
+ table1, t2 = update_from_fixture
+
+ mt = table1.alias()
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ u = update(
+ table1,
+ values={
+ table1.c.name: select(
+ [mt.c.name], mt.c.myid == table1.c.myid
+ )
+ },
+ )
+ self.assert_compile(
+ u,
+ "UPDATE mytable SET name=(SELECT mytable_1.name FROM "
+ "mytable AS mytable_1 WHERE "
+ "mytable_1.myid = mytable.myid)",
+ )
+
+ def test_correlated_update_three(self, update_from_fixture):
+ table1, table2 = update_from_fixture
+
+ # test against a regular constructed subquery
+ s = select([table2], table2.c.otherid == table1.c.myid)
+ with testing.expect_warnings(
+ "implicitly coercing SELECT object to scalar subquery"
+ ):
+ u = update(
+ table1, table1.c.name == "jack", values={table1.c.name: s}
+ )
+ self.assert_compile(
+ u,
+ "UPDATE mytable SET name=(SELECT myothertable.otherid, "
+ "myothertable.othername FROM myothertable WHERE "
+ "myothertable.otherid = mytable.myid) "
+ "WHERE mytable.name = :name_1",
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
