--- /dev/null
+.. change::
+ :tags: mssql, bug
+ :tickets: 7300
+
+ Adjusted the compiler's generation of "post compile" symbols including
+ those used for "expanding IN" as well as for the "schema translate map" to
+ not be based directly on plain bracketed strings with underscores, as this
+ conflicts directly with SQL Server's quoting format of also using brackets,
+ which produces false matches when the compiler replaces "post compile" and
+ "schema translate" symbols. The issue created easy to reproduce examples
+ both with the :meth:`.Inspector.get_schema_names` method when used in
+ conjunction with the
+ :paramref:`_engine.Connection.execution_options.schema_translate_map`
+ feature, as well in the unlikely case that a symbol overlapping with the
+ internal name "POSTCOMPILE" would be used with a feature like "expanding
+ in".
+
>>> print(column('x').in_([1, 2, 3]))
- x IN ([POSTCOMPILE_x_1])
+ x IN (__[POSTCOMPILE_x_1])
-The special bound form ``POSTCOMPILE`` is rendered into individual parameters
+The special bound form ``__[POSTCOMPILE`` is rendered into individual parameters
at execution time, illustrated below:
.. sourcecode:: pycon+sql
"NOT IN" is available via the :meth:`_sql.ColumnOperators.not_in` operator::
>>> print(column('x').not_in([1, 2, 3]))
- (x NOT IN ([POSTCOMPILE_x_1]))
+ (x NOT IN (__[POSTCOMPILE_x_1]))
This is typically more easily available by negating with the ``~`` operator::
>>> print(~column('x').in_([1, 2, 3]))
- (x NOT IN ([POSTCOMPILE_x_1]))
+ (x NOT IN (__[POSTCOMPILE_x_1]))
Tuple IN Expressions
~~~~~~~~~~~~~~~~~~~~
>>> tup = tuple_(column('x', Integer), column('y', Integer))
>>> expr = tup.in_([(1, 2), (3, 4)])
>>> print(expr)
- (x, y) IN ([POSTCOMPILE_param_1])
+ (x, y) IN (__[POSTCOMPILE_param_1])
To illustrate the parameters rendered:
# test/dialect/mssql/test_compiler.py -> test_schema_many_tokens_*
#
+ if schema.startswith("__[SCHEMA_"):
+ return None, schema
+
push = []
symbol = ""
bracket = False
return expr
statement = re.sub(
- r"\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]",
+ r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]",
process_expanding,
self.string,
)
# for postcompile w/ expanding, move the "wrapped" part
# of this into the inside
m = re.match(
- r"^(.*)\(\[POSTCOMPILE_(\S+?)\]\)(.*)$", wrapped
+ r"^(.*)\(__\[POSTCOMPILE_(\S+?)\]\)(.*)$", wrapped
)
- wrapped = "([POSTCOMPILE_%s~~%s~~REPL~~%s~~])" % (
+ wrapped = "(__[POSTCOMPILE_%s~~%s~~REPL~~%s~~])" % (
m.group(2),
m.group(1),
m.group(3),
self.escaped_bind_names = {}
self.escaped_bind_names[escaped_from] = name
if post_compile:
- return "[POSTCOMPILE_%s]" % name
+ return "__[POSTCOMPILE_%s]" % name
else:
return self.bindtemplate % {"name": name}
"in schema translate name '%s'" % name
)
return quoted_name(
- "[SCHEMA_%s]" % (name or "_none"), quote=False
+ "__[SCHEMA_%s]" % (name or "_none"), quote=False
)
else:
return obj.schema
)
return self.quote_schema(effective_schema)
- return re.sub(r"(\[SCHEMA_([^\]]+)\])", replace, statement)
+ return re.sub(r"(__\[SCHEMA_([^\]]+)\])", replace, statement)
def _escape_identifier(self, value):
"""Escape an identifier.
import re
import sqlalchemy as sa
-from sqlalchemy import func
from .. import config
from .. import engines
from .. import eq_
from ..schema import Table
from ... import event
from ... import ForeignKey
+from ... import func
from ... import Identity
from ... import inspect
from ... import Integer
from ...schema import DDL
from ...schema import Index
from ...sql.elements import quoted_name
+from ...sql.schema import BLANK_SCHEMA
from ...testing import is_false
from ...testing import is_true
self.assert_(testing.config.test_schema in insp.get_schema_names())
+ @testing.requires.schema_reflection
+ def test_get_schema_names_w_translate_map(self, connection):
+ """test #7300"""
+
+ connection = connection.execution_options(
+ schema_translate_map={
+ "foo": "bar",
+ BLANK_SCHEMA: testing.config.test_schema,
+ }
+ )
+ insp = inspect(connection)
+
+ self.assert_(testing.config.test_schema in insp.get_schema_names())
+
@testing.requires.schema_reflection
def test_dialect_initialize(self):
engine = engines.testing_engine()
self.assert_compile(
stmt,
"SELECT some_table.id FROM some_table "
- "WHERE some_table.x = [POSTCOMPILE_q]",
+ "WHERE some_table.x = __[POSTCOMPILE_q]",
{},
)
@testing.combinations(
(
lambda: select(literal("x"), literal("y")),
- "SELECT [POSTCOMPILE_param_1] AS anon_1, "
- "[POSTCOMPILE_param_2] AS anon_2",
+ "SELECT __[POSTCOMPILE_param_1] AS anon_1, "
+ "__[POSTCOMPILE_param_2] AS anon_2",
{
"check_literal_execute": {"param_1": "x", "param_2": "y"},
"check_post_param": {},
(
lambda t: select(t).where(t.c.foo.in_(["x", "y", "z"])),
"SELECT sometable.foo FROM sometable WHERE sometable.foo "
- "IN ([POSTCOMPILE_foo_1])",
+ "IN (__[POSTCOMPILE_foo_1])",
{
"check_literal_execute": {"foo_1": ["x", "y", "z"]},
"check_post_param": {},
crit = q.c.myid == table1.c.myid
self.assert_compile(
select("*").where(crit),
- "SELECT * FROM (SELECT TOP [POSTCOMPILE_param_1] mytable.myid AS "
+ "SELECT * FROM (SELECT TOP __[POSTCOMPILE_param_1] "
+ "mytable.myid AS "
"myid FROM mytable ORDER BY mytable.myid) AS foo, mytable WHERE "
"foo.myid = mytable.myid",
)
self.assert_compile(
u,
"SELECT t1.col3 AS col3, t1.col4 AS col4 "
- "FROM t1 WHERE t1.col2 IN ([POSTCOMPILE_col2_1]) "
+ "FROM t1 WHERE t1.col2 IN (__[POSTCOMPILE_col2_1]) "
"UNION SELECT t2.col3 AS col3, "
"t2.col4 AS col4 FROM t2 WHERE t2.col2 IN "
- "([POSTCOMPILE_col2_2]) ORDER BY col3, col4",
+ "(__[POSTCOMPILE_col2_2]) ORDER BY col3, col4",
checkparams={
"col2_1": ["t1col2r1", "t1col2r2"],
"col2_2": ["t2col2r2", "t2col2r3"],
u.alias("bar").select(),
"SELECT bar.col3, bar.col4 FROM (SELECT "
"t1.col3 AS col3, t1.col4 AS col4 FROM t1 "
- "WHERE t1.col2 IN ([POSTCOMPILE_col2_1]) UNION "
+ "WHERE t1.col2 IN (__[POSTCOMPILE_col2_1]) UNION "
"SELECT t2.col3 AS col3, t2.col4 AS col4 "
- "FROM t2 WHERE t2.col2 IN ([POSTCOMPILE_col2_2])) AS bar",
+ "FROM t2 WHERE t2.col2 IN (__[POSTCOMPILE_col2_2])) AS bar",
checkparams={
"col2_1": ["t1col2r1", "t1col2r2"],
"col2_2": ["t2col2r2", "t2col2r3"],
self.assert_compile(
s,
- "SELECT TOP [POSTCOMPILE_param_1] t.x, t.y FROM t "
+ "SELECT TOP __[POSTCOMPILE_param_1] t.x, t.y FROM t "
"WHERE t.x = :x_1 ORDER BY t.y",
checkparams={"x_1": 5, "param_1": 10},
)
self.assert_compile(
s,
- "SELECT TOP [POSTCOMPILE_param_1] t.x, t.y FROM t "
+ "SELECT TOP __[POSTCOMPILE_param_1] t.x, t.y FROM t "
"WHERE t.x = :x_1 ORDER BY t.y",
checkparams={"x_1": 5, "param_1": 0},
)
# of zero, so produces TOP 0
self.assert_compile(
s,
- "SELECT TOP [POSTCOMPILE_param_1] t.x, t.y FROM t "
+ "SELECT TOP __[POSTCOMPILE_param_1] t.x, t.y FROM t "
"WHERE t.x = :x_1 ORDER BY t.y",
checkparams={"x_1": 5, "param_1": 0},
)
5,
0,
{"percent": True},
- "TOP [POSTCOMPILE_param_1] PERCENT",
+ "TOP __[POSTCOMPILE_param_1] PERCENT",
{"param_1": 5},
),
(
5,
None,
{"percent": True, "with_ties": True},
- "TOP [POSTCOMPILE_param_1] PERCENT WITH TIES",
+ "TOP __[POSTCOMPILE_param_1] PERCENT WITH TIES",
{"param_1": 5},
),
(
5,
0,
{"with_ties": True},
- "TOP [POSTCOMPILE_param_1] WITH TIES",
+ "TOP __[POSTCOMPILE_param_1] WITH TIES",
{"param_1": 5},
),
(
5,
0,
{"percent": True},
- "TOP [POSTCOMPILE_param_1] PERCENT",
+ "TOP __[POSTCOMPILE_param_1] PERCENT",
{"param_1": 5},
),
(
5,
None,
{"percent": True, "with_ties": True},
- "TOP [POSTCOMPILE_param_1] PERCENT WITH TIES",
+ "TOP __[POSTCOMPILE_param_1] PERCENT WITH TIES",
{"param_1": 5},
),
(
5,
0,
{"with_ties": True},
- "TOP [POSTCOMPILE_param_1] WITH TIES",
+ "TOP __[POSTCOMPILE_param_1] WITH TIES",
{"param_1": 5},
),
(
r = connection.execute(t1.insert(), dict(descr="hello"))
eq_(r.inserted_primary_key, (100,))
+ def test_compiler_symbol_conflict(self, connection, metadata):
+ t = Table("t", metadata, Column("POSTCOMPILE_DATA", String(50)))
+
+ t.create(connection)
+
+ connection.execute(t.insert().values(POSTCOMPILE_DATA="some data"))
+ eq_(
+ connection.scalar(
+ select(t.c.POSTCOMPILE_DATA).where(
+ t.c.POSTCOMPILE_DATA.in_(["some data", "some other data"])
+ )
+ ),
+ "some data",
+ )
+
@testing.provide_metadata
def _test_disable_scope_identity(self):
engine = engines.testing_engine(options={"use_scope_identity": False})
def test_bindparam_quote_works_on_expanding(self):
self.assert_compile(
bindparam("uid", expanding=True),
- "([POSTCOMPILE_uid])",
+ "(__[POSTCOMPILE_uid])",
dialect=cx_oracle.dialect(),
)
"anon_2.col2 AS col2, ROWNUM AS ora_rn FROM (SELECT "
"sometable.col1 AS col1, sometable.col2 AS "
"col2 FROM sometable) anon_2 WHERE ROWNUM <= "
- "[POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) anon_1 "
+ "__[POSTCOMPILE_param_1] + __[POSTCOMPILE_param_2]) anon_1 "
"WHERE ora_rn > "
- "[POSTCOMPILE_param_2]",
+ "__[POSTCOMPILE_param_2]",
checkparams={"param_1": 10, "param_2": 20},
)
self.assert_compile(
s,
"SELECT anon_1.col1, anon_1.col2 FROM "
- "(SELECT /*+ FIRST_ROWS([POSTCOMPILE_param_1]) */ "
+ "(SELECT /*+ FIRST_ROWS(__[POSTCOMPILE_param_1]) */ "
"anon_2.col1 AS col1, "
"anon_2.col2 AS col2, ROWNUM AS ora_rn FROM (SELECT "
"sometable.col1 AS col1, sometable.col2 AS "
"col2 FROM sometable) anon_2 WHERE ROWNUM <= "
- "[POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) anon_1 "
+ "__[POSTCOMPILE_param_1] + __[POSTCOMPILE_param_2]) anon_1 "
"WHERE ora_rn > "
- "[POSTCOMPILE_param_2]",
+ "__[POSTCOMPILE_param_2]",
checkparams={"param_1": 10, "param_2": 20},
dialect=oracle.OracleDialect(optimize_limits=True),
)
"ROWNUM AS ora_rn "
"FROM (SELECT sometable.col1 AS col1, "
"sometable.col2 AS col2 FROM sometable) anon_3 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) "
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] + "
+ "__[POSTCOMPILE_param_2]) "
"anon_2 "
- "WHERE ora_rn > [POSTCOMPILE_param_2]) anon_1",
+ "WHERE ora_rn > __[POSTCOMPILE_param_2]) anon_1",
checkparams={"param_1": 10, "param_2": 20},
)
"ROWNUM AS ora_rn "
"FROM (SELECT sometable.col1 AS col1, "
"sometable.col2 AS col2 FROM sometable) anon_3 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) "
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] + "
+ "__[POSTCOMPILE_param_2]) "
"anon_2 "
- "WHERE ora_rn > [POSTCOMPILE_param_2]) anon_1",
+ "WHERE ora_rn > __[POSTCOMPILE_param_2]) anon_1",
)
c = s2.compile(dialect=oracle.OracleDialect())
eq_(len(c._result_columns), 2)
"sometable.col1 AS col1, sometable.col2 AS "
"col2 FROM sometable ORDER BY "
"sometable.col2) anon_2 WHERE ROWNUM <= "
- "[POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_2]",
+ "__[POSTCOMPILE_param_1] + __[POSTCOMPILE_param_2]) anon_1 "
+ "WHERE ora_rn > __[POSTCOMPILE_param_2]",
checkparams={"param_1": 10, "param_2": 20},
)
c = s.compile(dialect=oracle.OracleDialect())
"SELECT anon_1.col1, anon_1.col2 FROM (SELECT "
"sometable.col1 AS col1, sometable.col2 AS "
"col2 FROM sometable ORDER BY "
- "sometable.col2) anon_1 WHERE ROWNUM <= [POSTCOMPILE_param_1] "
+ "sometable.col2) anon_1 WHERE ROWNUM <= __[POSTCOMPILE_param_1] "
"FOR UPDATE",
checkparams={"param_1": 10},
)
s = select(t).with_for_update().limit(10).order_by(t.c.col2)
self.assert_compile(
s,
- "SELECT /*+ FIRST_ROWS([POSTCOMPILE_param_1]) */ "
+ "SELECT /*+ FIRST_ROWS(__[POSTCOMPILE_param_1]) */ "
"anon_1.col1, anon_1.col2 FROM (SELECT "
"sometable.col1 AS col1, sometable.col2 AS "
"col2 FROM sometable ORDER BY "
- "sometable.col2) anon_1 WHERE ROWNUM <= [POSTCOMPILE_param_1] "
+ "sometable.col2) anon_1 WHERE ROWNUM <= __[POSTCOMPILE_param_1] "
"FOR UPDATE",
checkparams={"param_1": 10},
dialect=oracle.OracleDialect(optimize_limits=True),
"sometable.col1 AS col1, sometable.col2 AS "
"col2 FROM sometable ORDER BY "
"sometable.col2) anon_2 WHERE ROWNUM <= "
- "[POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_2] FOR "
+ "__[POSTCOMPILE_param_1] + __[POSTCOMPILE_param_2]) anon_1 "
+ "WHERE ora_rn > __[POSTCOMPILE_param_2] FOR "
"UPDATE",
checkparams={"param_1": 10, "param_2": 20},
)
"col1, anon_2.col2 AS col2, ROWNUM AS ora_rn FROM "
"(SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
"FROM sometable ORDER BY sometable.col2) anon_2 WHERE "
- "ROWNUM <= [POSTCOMPILE_param_1] + :param_2 + :param_3) anon_1 "
+ "ROWNUM <= __[POSTCOMPILE_param_1] + :param_2 + :param_3) anon_1 "
"WHERE ora_rn > :param_2 + :param_3",
checkparams={"param_1": 10, "param_2": 10, "param_3": 20},
)
'SELECT anon_1."SUM(ABC)" FROM '
'(SELECT SUM(ABC) AS "SUM(ABC)" '
"FROM my_table ORDER BY SUM(ABC)) anon_1 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1]",
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1]",
)
col = literal_column("SUM(ABC)").label(quoted_name("SUM(ABC)", True))
'SELECT anon_1."SUM(ABC)" FROM '
'(SELECT SUM(ABC) AS "SUM(ABC)" '
"FROM my_table ORDER BY SUM(ABC)) anon_1 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1]",
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1]",
)
col = literal_column("SUM(ABC)").label("SUM(ABC)_")
'SELECT anon_1."SUM(ABC)_" FROM '
'(SELECT SUM(ABC) AS "SUM(ABC)_" '
"FROM my_table ORDER BY SUM(ABC)) anon_1 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1]",
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1]",
)
col = literal_column("SUM(ABC)").label(quoted_name("SUM(ABC)_", True))
'SELECT anon_1."SUM(ABC)_" FROM '
'(SELECT SUM(ABC) AS "SUM(ABC)_" '
"FROM my_table ORDER BY SUM(ABC)) anon_1 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1]",
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1]",
)
def test_for_update(self):
"SELECT anon_1.myid, anon_1.name FROM "
"(SELECT mytable.myid AS myid, mytable.name AS name "
"FROM mytable WHERE mytable.myid = :myid_1) anon_1 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] "
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] "
"FOR UPDATE OF anon_1.name NOWAIT",
checkparams={"param_1": 10, "myid_1": 7},
)
"SELECT anon_1.myid FROM "
"(SELECT mytable.myid AS myid, mytable.name AS name "
"FROM mytable WHERE mytable.myid = :myid_1) anon_1 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] "
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] "
"FOR UPDATE OF anon_1.name NOWAIT",
)
"ROWNUM AS ora_rn "
"FROM (SELECT mytable.myid AS myid, mytable.name AS name "
"FROM mytable WHERE mytable.myid = :myid_1) anon_2 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) "
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] + "
+ "__[POSTCOMPILE_param_2]) "
"anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_2] "
+ "WHERE ora_rn > __[POSTCOMPILE_param_2] "
"FOR UPDATE OF anon_1.name NOWAIT",
checkparams={"param_1": 10, "param_2": 50, "myid_1": 7},
)
"FROM (SELECT mytable.myid AS myid, mytable.name AS name "
"FROM mytable WHERE mytable.myid = :myid_1) anon_2 "
"WHERE "
- "ROWNUM <= [POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_2] "
+ "ROWNUM <= __[POSTCOMPILE_param_1] + "
+ "__[POSTCOMPILE_param_2]) anon_1 "
+ "WHERE ora_rn > __[POSTCOMPILE_param_2] "
"FOR UPDATE OF anon_1.name NOWAIT",
checkparams={"param_1": 10, "param_2": 50, "myid_1": 7},
)
"mytable.bar AS bar, "
"mytable.foo AS foo FROM mytable "
"WHERE mytable.myid = :myid_1) anon_2 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) "
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] + "
+ "__[POSTCOMPILE_param_2]) "
"anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_2] "
+ "WHERE ora_rn > __[POSTCOMPILE_param_2] "
"FOR UPDATE OF anon_1.foo, anon_1.bar NOWAIT",
checkparams={"param_1": 10, "param_2": 50, "myid_1": 7},
)
"SELECT anon_1.col1, anon_1.col2 FROM "
"(SELECT sometable.col1 AS col1, "
"sometable.col2 AS col2 FROM sometable) anon_1 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1]",
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1]",
dialect=dialect,
)
"anon_2.col1 AS col1, anon_2.col2 AS col2, ROWNUM AS ora_rn "
"FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
"FROM sometable) anon_2) anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_1]",
+ "WHERE ora_rn > __[POSTCOMPILE_param_1]",
dialect=dialect,
)
"anon_2.col1 AS col1, anon_2.col2 AS col2, ROWNUM AS ora_rn "
"FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
"FROM sometable) anon_2 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] + "
- "[POSTCOMPILE_param_2]) anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_2]",
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] + "
+ "__[POSTCOMPILE_param_2]) anon_1 "
+ "WHERE ora_rn > __[POSTCOMPILE_param_2]",
dialect=dialect,
)
"SELECT anon_1.col1, anon_1.col2 FROM "
"(SELECT sometable.col1 AS col1, "
"sometable.col2 AS col2 FROM sometable) anon_1 WHERE ROWNUM "
- "<= [POSTCOMPILE_param_1]",
+ "<= __[POSTCOMPILE_param_1]",
dialect=dialect,
)
"ROWNUM AS ora_rn "
"FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
"FROM sometable) anon_2) anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_1]",
+ "WHERE ora_rn > __[POSTCOMPILE_param_1]",
dialect=dialect,
)
"ROWNUM AS ora_rn "
"FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
"FROM sometable) anon_2 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] + "
- "[POSTCOMPILE_param_2]) anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_2]",
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] + "
+ "__[POSTCOMPILE_param_2]) anon_1 "
+ "WHERE ora_rn > __[POSTCOMPILE_param_2]",
dialect=dialect,
checkparams={"param_1": 10, "param_2": 10},
)
"thirdtable.userid(+) = "
"myothertable.otherid AND mytable.myid = "
"myothertable.otherid ORDER BY mytable.name) anon_2 "
- "WHERE ROWNUM <= [POSTCOMPILE_param_1] + [POSTCOMPILE_param_2]) "
+ "WHERE ROWNUM <= __[POSTCOMPILE_param_1] + "
+ "__[POSTCOMPILE_param_2]) "
"anon_1 "
- "WHERE ora_rn > [POSTCOMPILE_param_2]",
+ "WHERE ora_rn > __[POSTCOMPILE_param_2]",
checkparams={"param_1": 10, "param_2": 5},
dialect=oracle.dialect(use_ansi=False),
)
self.assert_compile(
expr,
- "x IN ([POSTCOMPILE_x_1~~~~REPL~~::myenum[]~~])",
+ "x IN (__[POSTCOMPILE_x_1~~~~REPL~~::myenum[]~~])",
dialect=postgresql.psycopg2.dialect(),
)
self.assert_compile(
expr,
- "x IN ([POSTCOMPILE_x_1~~~~REPL~~::VARCHAR(15)[]~~])",
+ "x IN (__[POSTCOMPILE_x_1~~~~REPL~~::VARCHAR(15)[]~~])",
dialect=postgresql.psycopg2.dialect(),
)
.in_([(1, 2), (3, 4)])
.compile(dialect=sqlite.dialect())
)
- eq_(str(compiled), "(q, p) IN ([POSTCOMPILE_param_1])")
+ eq_(str(compiled), "(q, p) IN (__[POSTCOMPILE_param_1])")
eq_(
compiled._literal_execute_expanding_parameter(
"param_1",
t1.drop(conn)
asserter.assert_(
- CompiledSQL("CREATE TABLE [SCHEMA__none].t1 (x INTEGER)"),
- CompiledSQL("CREATE TABLE [SCHEMA_foo].t2 (x INTEGER)"),
- CompiledSQL("CREATE TABLE [SCHEMA_bar].t3 (x INTEGER)"),
- CompiledSQL("DROP TABLE [SCHEMA_bar].t3"),
- CompiledSQL("DROP TABLE [SCHEMA_foo].t2"),
- CompiledSQL("DROP TABLE [SCHEMA__none].t1"),
+ CompiledSQL("CREATE TABLE __[SCHEMA__none].t1 (x INTEGER)"),
+ CompiledSQL("CREATE TABLE __[SCHEMA_foo].t2 (x INTEGER)"),
+ CompiledSQL("CREATE TABLE __[SCHEMA_bar].t3 (x INTEGER)"),
+ CompiledSQL("DROP TABLE __[SCHEMA_bar].t3"),
+ CompiledSQL("DROP TABLE __[SCHEMA_foo].t2"),
+ CompiledSQL("DROP TABLE __[SCHEMA__none].t1"),
)
def test_ddl_hastable(self, plain_tables, connection):
conn.execute(t3.delete(), execution_options=execution_options)
asserter.assert_(
- CompiledSQL("INSERT INTO [SCHEMA__none].t1 (x) VALUES (:x)"),
- CompiledSQL("INSERT INTO [SCHEMA_foo].t2 (x) VALUES (:x)"),
- CompiledSQL("INSERT INTO [SCHEMA_bar].t3 (x) VALUES (:x)"),
+ CompiledSQL("INSERT INTO __[SCHEMA__none].t1 (x) VALUES (:x)"),
+ CompiledSQL("INSERT INTO __[SCHEMA_foo].t2 (x) VALUES (:x)"),
+ CompiledSQL("INSERT INTO __[SCHEMA_bar].t3 (x) VALUES (:x)"),
CompiledSQL(
- "UPDATE [SCHEMA__none].t1 SET x=:x WHERE "
- "[SCHEMA__none].t1.x = :x_1"
+ "UPDATE __[SCHEMA__none].t1 SET x=:x WHERE "
+ "__[SCHEMA__none].t1.x = :x_1"
),
CompiledSQL(
- "UPDATE [SCHEMA_foo].t2 SET x=:x WHERE "
- "[SCHEMA_foo].t2.x = :x_1"
+ "UPDATE __[SCHEMA_foo].t2 SET x=:x WHERE "
+ "__[SCHEMA_foo].t2.x = :x_1"
),
CompiledSQL(
- "UPDATE [SCHEMA_bar].t3 SET x=:x WHERE "
- "[SCHEMA_bar].t3.x = :x_1"
+ "UPDATE __[SCHEMA_bar].t3 SET x=:x WHERE "
+ "__[SCHEMA_bar].t3.x = :x_1"
),
- CompiledSQL("SELECT [SCHEMA__none].t1.x FROM [SCHEMA__none].t1"),
- CompiledSQL("SELECT [SCHEMA_foo].t2.x FROM [SCHEMA_foo].t2"),
- CompiledSQL("SELECT [SCHEMA_bar].t3.x FROM [SCHEMA_bar].t3"),
- CompiledSQL("DELETE FROM [SCHEMA__none].t1"),
- CompiledSQL("DELETE FROM [SCHEMA_foo].t2"),
- CompiledSQL("DELETE FROM [SCHEMA_bar].t3"),
+ CompiledSQL(
+ "SELECT __[SCHEMA__none].t1.x FROM __[SCHEMA__none].t1"
+ ),
+ CompiledSQL("SELECT __[SCHEMA_foo].t2.x FROM __[SCHEMA_foo].t2"),
+ CompiledSQL("SELECT __[SCHEMA_bar].t3.x FROM __[SCHEMA_bar].t3"),
+ CompiledSQL("DELETE FROM __[SCHEMA__none].t1"),
+ CompiledSQL("DELETE FROM __[SCHEMA_foo].t2"),
+ CompiledSQL("DELETE FROM __[SCHEMA_bar].t3"),
)
def test_crud(self, plain_tables, connection):
conn.execute(t3.delete())
asserter.assert_(
- CompiledSQL("INSERT INTO [SCHEMA__none].t1 (x) VALUES (:x)"),
- CompiledSQL("INSERT INTO [SCHEMA_foo].t2 (x) VALUES (:x)"),
- CompiledSQL("INSERT INTO [SCHEMA_bar].t3 (x) VALUES (:x)"),
+ CompiledSQL("INSERT INTO __[SCHEMA__none].t1 (x) VALUES (:x)"),
+ CompiledSQL("INSERT INTO __[SCHEMA_foo].t2 (x) VALUES (:x)"),
+ CompiledSQL("INSERT INTO __[SCHEMA_bar].t3 (x) VALUES (:x)"),
+ CompiledSQL(
+ "UPDATE __[SCHEMA__none].t1 SET x=:x WHERE "
+ "__[SCHEMA__none].t1.x = :x_1"
+ ),
CompiledSQL(
- "UPDATE [SCHEMA__none].t1 SET x=:x WHERE "
- "[SCHEMA__none].t1.x = :x_1"
+ "UPDATE __[SCHEMA_foo].t2 SET x=:x WHERE "
+ "__[SCHEMA_foo].t2.x = :x_1"
),
CompiledSQL(
- "UPDATE [SCHEMA_foo].t2 SET x=:x WHERE "
- "[SCHEMA_foo].t2.x = :x_1"
+ "UPDATE __[SCHEMA_bar].t3 SET x=:x WHERE "
+ "__[SCHEMA_bar].t3.x = :x_1"
),
CompiledSQL(
- "UPDATE [SCHEMA_bar].t3 SET x=:x WHERE "
- "[SCHEMA_bar].t3.x = :x_1"
+ "SELECT __[SCHEMA__none].t1.x FROM __[SCHEMA__none].t1"
),
- CompiledSQL("SELECT [SCHEMA__none].t1.x FROM [SCHEMA__none].t1"),
- CompiledSQL("SELECT [SCHEMA_foo].t2.x FROM [SCHEMA_foo].t2"),
- CompiledSQL("SELECT [SCHEMA_bar].t3.x FROM [SCHEMA_bar].t3"),
- CompiledSQL("DELETE FROM [SCHEMA__none].t1"),
- CompiledSQL("DELETE FROM [SCHEMA_foo].t2"),
- CompiledSQL("DELETE FROM [SCHEMA_bar].t3"),
+ CompiledSQL("SELECT __[SCHEMA_foo].t2.x FROM __[SCHEMA_foo].t2"),
+ CompiledSQL("SELECT __[SCHEMA_bar].t3.x FROM __[SCHEMA_bar].t3"),
+ CompiledSQL("DELETE FROM __[SCHEMA__none].t1"),
+ CompiledSQL("DELETE FROM __[SCHEMA_foo].t2"),
+ CompiledSQL("DELETE FROM __[SCHEMA_bar].t3"),
)
def test_via_engine(self, plain_tables, metadata):
with eng.connect() as conn:
conn.execute(select(t2.c.x))
asserter.assert_(
- CompiledSQL("SELECT [SCHEMA_foo].t2.x FROM [SCHEMA_foo].t2")
+ CompiledSQL("SELECT __[SCHEMA_foo].t2.x FROM __[SCHEMA_foo].t2")
)
self.assert_compile(
s.query(Derived.data_syn).filter(Derived.data_syn == "foo"),
"SELECT test.data AS test_data FROM test WHERE test.data = "
- ":data_1 AND test.type IN ([POSTCOMPILE_type_1])",
+ ":data_1 AND test.type IN (__[POSTCOMPILE_type_1])",
dialect="default",
checkparams={"type_1": ["derived"], "data_1": "foo"},
)
"companies.name AS companies_name FROM companies "
"LEFT OUTER JOIN employees AS employees_1 ON "
"companies.company_id = employees_1.company_id "
- "AND employees_1.type IN ([POSTCOMPILE_type_1])",
+ "AND employees_1.type IN (__[POSTCOMPILE_type_1])",
)
"engineer.engineer_info AS engineer_engineer_info, "
"engineer.manager_id AS engineer_manager_id "
"FROM employee JOIN engineer ON employee.id = engineer.id) "
- "AS anon_1 WHERE anon_1.employee_type IN ([POSTCOMPILE_type_1])",
+ "AS anon_1 WHERE anon_1.employee_type IN (__[POSTCOMPILE_type_1])",
)
"a.type AS a_type, "
"asub.asubdata AS asub_asubdata FROM a JOIN asub "
"ON a.id = asub.id "
- "WHERE a.id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE a.id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY a.id",
{"primary_keys": [2]},
),
"SELECT c.a_sub_id AS c_a_sub_id, "
"c.id AS c_id "
"FROM c WHERE c.a_sub_id "
- "IN ([POSTCOMPILE_primary_keys])",
+ "IN (__[POSTCOMPILE_primary_keys])",
{"primary_keys": [2]},
),
),
CompiledSQL(
"SELECT b.a_id AS b_a_id, b.id AS b_id FROM b "
- "WHERE b.a_id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE b.a_id IN (__[POSTCOMPILE_primary_keys])",
{"primary_keys": [1, 2]},
),
),
"engineers.primary_language AS engineers_primary_language "
"FROM people JOIN engineers "
"ON people.person_id = engineers.person_id "
- "WHERE people.person_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE people.person_id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY people.person_id",
{"primary_keys": [1, 2, 5]},
),
"managers.manager_name AS managers_manager_name "
"FROM people JOIN managers "
"ON people.person_id = managers.person_id "
- "WHERE people.person_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE people.person_id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY people.person_id",
{"primary_keys": [3, 4]},
),
"people.person_id AS people_person_id, "
"people.name AS people_name, people.type AS people_type "
"FROM people WHERE people.company_id "
- "IN ([POSTCOMPILE_primary_keys]) "
+ "IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY people.person_id",
{"primary_keys": [1, 2]},
),
"managers.manager_name AS managers_manager_name "
"FROM people JOIN managers "
"ON people.person_id = managers.person_id "
- "WHERE people.person_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE people.person_id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY people.person_id",
{"primary_keys": [3, 4]},
),
"engineers.primary_language AS engineers_primary_language "
"FROM people JOIN engineers "
"ON people.person_id = engineers.person_id "
- "WHERE people.person_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE people.person_id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY people.person_id",
{"primary_keys": [1, 2, 5]},
),
"c.c_data AS c_c_data, c.e_data AS c_e_data, "
"c.d_data AS c_d_data "
"FROM a JOIN c ON a.id = c.id "
- "WHERE a.id IN ([POSTCOMPILE_primary_keys]) ORDER BY a.id",
+ "WHERE a.id IN (__[POSTCOMPILE_primary_keys]) "
+ "ORDER BY a.id",
[{"primary_keys": [1, 2]}],
),
CompiledSQL(
"c.c_data AS c_c_data, "
"c.d_data AS c_d_data, c.e_data AS c_e_data "
"FROM a JOIN c ON a.id = c.id "
- "WHERE a.id IN ([POSTCOMPILE_primary_keys]) ORDER BY a.id",
+ "WHERE a.id IN (__[POSTCOMPILE_primary_keys]) "
+ "ORDER BY a.id",
[{"primary_keys": [1, 2]}],
),
),
"c.c_data AS c_c_data, c.e_data AS c_e_data, "
"c.d_data AS c_d_data "
"FROM a JOIN c ON a.id = c.id "
- "WHERE a.id IN ([POSTCOMPILE_primary_keys]) ORDER BY a.id",
+ "WHERE a.id IN (__[POSTCOMPILE_primary_keys]) "
+ "ORDER BY a.id",
[{"primary_keys": [1, 2]}],
),
CompiledSQL(
"c.c_data AS c_c_data, c.d_data AS c_d_data, "
"c.e_data AS c_e_data "
"FROM a JOIN c ON a.id = c.id "
- "WHERE a.id IN ([POSTCOMPILE_primary_keys]) ORDER BY a.id",
+ "WHERE a.id IN (__[POSTCOMPILE_primary_keys]) "
+ "ORDER BY a.id",
[{"primary_keys": [1, 2]}],
),
),
"e.id AS e_id, e.e_data AS e_e_data FROM a JOIN c "
"ON a.id = c.id LEFT OUTER JOIN d ON c.id = d.id "
"LEFT OUTER JOIN e ON c.id = e.id) AS poly "
- "WHERE poly.a_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE poly.a_id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY poly.a_id",
[{"primary_keys": [1, 2]}],
),
"e.e_data AS e_e_data FROM a JOIN c ON a.id = c.id "
"LEFT OUTER JOIN d ON c.id = d.id "
"LEFT OUTER JOIN e ON c.id = e.id) AS poly "
- "WHERE poly.a_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE poly.a_id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY poly.a_id",
[{"primary_keys": [1, 2]}],
),
"child.type AS child_type "
"FROM child JOIN child_subclass1 "
"ON child.id = child_subclass1.id "
- "WHERE child.id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE child.id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY child.id",
[{"primary_keys": [1]}],
),
"ON child.id = child_subclass1.id "
"LEFT OUTER JOIN other AS other_1 "
"ON child_subclass1.id = other_1.child_subclass_id "
- "WHERE child.id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE child.id IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY child.id",
[{"primary_keys": [1]}],
),
"a_b.kind AS a_b_kind, a_b.a_id AS a_b_a_id, a.id AS a_id_1, "
"a.kind AS a_kind, a.a_id AS a_a_id FROM a "
"LEFT OUTER JOIN a AS a_b ON a.id = a_b.a_id AND a_b.kind IN "
- "([POSTCOMPILE_kind_1]) LEFT OUTER JOIN x AS b_x "
+ "(__[POSTCOMPILE_kind_1]) LEFT OUTER JOIN x AS b_x "
"ON a_b.id = b_x.a_id",
)
joinedload(cls.links).joinedload(Link.child).joinedload(cls.links)
)
if cls is self.classes.Sub1:
- extra = " WHERE parent.type IN ([POSTCOMPILE_type_1])"
+ extra = " WHERE parent.type IN (__[POSTCOMPILE_type_1])"
else:
extra = ""
)
if Link.child.property.mapper.class_ is self.classes.Sub1:
- extra = "AND parent_1.type IN ([POSTCOMPILE_type_1]) "
+ extra = "AND parent_1.type IN (__[POSTCOMPILE_type_1]) "
else:
extra = ""
self.assert_compile(
select(subq1, subq2),
"SELECT (SELECT employees.employee_id FROM employees "
- "WHERE employees.type IN ([POSTCOMPILE_type_1])) AS foo, "
+ "WHERE employees.type IN (__[POSTCOMPILE_type_1])) AS foo, "
"(SELECT employees.employee_id FROM employees "
- "WHERE employees.type IN ([POSTCOMPILE_type_1])) AS bar",
+ "WHERE employees.type IN (__[POSTCOMPILE_type_1])) AS bar",
)
def test_multi_qualification(self):
"employees.engineer_info AS "
"employees_engineer_info, employees.type "
"AS employees_type FROM employees WHERE "
- "employees.type IN ([POSTCOMPILE_type_1])) AS "
+ "employees.type IN (__[POSTCOMPILE_type_1])) AS "
"anon_1",
use_default_dialect=True,
)
"employees.engineer_info AS "
"employees_engineer_info, employees.type "
"AS employees_type FROM employees WHERE "
- "employees.type IN ([POSTCOMPILE_type_1])) AS "
- "anon_1 WHERE anon_1.employees_type IN ([POSTCOMPILE_type_2])",
+ "employees.type IN (__[POSTCOMPILE_type_1])) AS "
+ "anon_1 WHERE anon_1.employees_type IN (__[POSTCOMPILE_type_2])",
use_default_dialect=True,
)
sess.query(a1.employee_id).select_from(a1),
"SELECT employees_1.employee_id AS employees_1_employee_id "
"FROM employees AS employees_1 WHERE employees_1.type "
- "IN ([POSTCOMPILE_type_1])",
+ "IN (__[POSTCOMPILE_type_1])",
)
self.assert_compile(
sess.query(literal("1")).select_from(a1),
"SELECT :param_1 AS anon_1 FROM employees AS employees_1 "
- "WHERE employees_1.type IN ([POSTCOMPILE_type_1])",
+ "WHERE employees_1.type IN (__[POSTCOMPILE_type_1])",
)
def test_from_statement_select(self):
"SELECT employees.employee_id, employees.name, "
"employees.manager_data, employees.engineer_info, "
"employees.type FROM employees WHERE employees.type "
- "IN ([POSTCOMPILE_type_1])",
+ "IN (__[POSTCOMPILE_type_1])",
)
def test_from_statement_update(self):
self.assert_compile(
q,
"UPDATE employees SET engineer_info=:engineer_info "
- "WHERE employees.type IN ([POSTCOMPILE_type_1]) "
+ "WHERE employees.type IN (__[POSTCOMPILE_type_1]) "
"RETURNING employees.employee_id",
dialect="default_enhanced",
)
"employees.engineer_info AS employees_engineer_info, "
"employees.type AS employees_type FROM employees "
"WHERE employees.engineer_info = :engineer_info_1 "
- "AND employees.type IN ([POSTCOMPILE_type_1]) "
+ "AND employees.type IN (__[POSTCOMPILE_type_1]) "
"%(token)s "
"SELECT employees.employee_id AS employees_employee_id, "
"employees.name AS employees_name, "
"employees.engineer_info AS employees_engineer_info, "
"employees.type AS employees_type FROM employees "
"WHERE employees.manager_data = :manager_data_1 "
- "AND employees.type IN ([POSTCOMPILE_type_2])) AS anon_1"
+ "AND employees.type IN (__[POSTCOMPILE_type_2])) AS anon_1"
)
for meth, token in [
"employees.name AS employees_name, employees.manager_data "
"AS employees_manager_data, employees.engineer_info "
"AS employees_engineer_info, employees.type AS employees_type "
- "FROM employees WHERE employees.type IN ([POSTCOMPILE_type_1]) "
+ "FROM employees WHERE employees.type IN (__[POSTCOMPILE_type_1]) "
"GROUP BY employees.employee_id HAVING employees.name = :name_1",
)
"SELECT count(*) AS count_1 "
"FROM (SELECT employees.employee_id AS employees_employee_id "
"FROM employees "
- "WHERE employees.type IN ([POSTCOMPILE_type_1])) AS anon_1",
+ "WHERE employees.type IN (__[POSTCOMPILE_type_1])) AS anon_1",
use_default_dialect=True,
)
),
"SELECT EXISTS (SELECT 1 FROM employees WHERE "
"employees.name = :name_1 AND employees.type "
- "IN ([POSTCOMPILE_type_1])) AS anon_1",
+ "IN (__[POSTCOMPILE_type_1])) AS anon_1",
)
def test_type_filtering(self):
CompiledSQL(
"SELECT employee.id AS employee_id, employee.name AS "
"employee_name, employee.type AS employee_type "
- "FROM employee WHERE employee.type IN ([POSTCOMPILE_type_1])",
+ "FROM employee WHERE employee.type IN "
+ "(__[POSTCOMPILE_type_1])",
params=[{"type_1": ["manager"]}],
),
CompiledSQL(
"employee_stuff_name, anon_1.employee_id "
"AS anon_1_employee_id FROM (SELECT "
"employee.id AS employee_id FROM employee "
- "WHERE employee.type IN ([POSTCOMPILE_type_1])) AS anon_1 "
+ "WHERE employee.type IN (__[POSTCOMPILE_type_1])) AS anon_1 "
"JOIN employee_stuff ON anon_1.employee_id "
"= employee_stuff.employee_id",
params=[{"type_1": ["manager"]}],
"employees.name AS employees_name "
"FROM companies LEFT OUTER JOIN employees ON companies.company_id "
"= employees.company_id "
- "AND employees.type IN ([POSTCOMPILE_type_1])",
+ "AND employees.type IN (__[POSTCOMPILE_type_1])",
)
def test_outer_join_prop_alias(self):
"employees_1_name FROM companies LEFT OUTER "
"JOIN employees AS employees_1 ON companies.company_id "
"= employees_1.company_id "
- "AND employees_1.type IN ([POSTCOMPILE_type_1])",
+ "AND employees_1.type IN (__[POSTCOMPILE_type_1])",
)
def test_outer_join_literal_onclause(self):
"employees.company_id AS employees_company_id FROM companies "
"LEFT OUTER JOIN employees ON "
"companies.company_id = employees.company_id "
- "AND employees.type IN ([POSTCOMPILE_type_1])",
+ "AND employees.type IN (__[POSTCOMPILE_type_1])",
)
def test_outer_join_literal_onclause_alias(self):
"employees_1.company_id AS employees_1_company_id "
"FROM companies LEFT OUTER JOIN employees AS employees_1 ON "
"companies.company_id = employees_1.company_id "
- "AND employees_1.type IN ([POSTCOMPILE_type_1])",
+ "AND employees_1.type IN (__[POSTCOMPILE_type_1])",
)
def test_outer_join_no_onclause(self):
"employees.company_id AS employees_company_id "
"FROM companies LEFT OUTER JOIN employees ON "
"companies.company_id = employees.company_id "
- "AND employees.type IN ([POSTCOMPILE_type_1])",
+ "AND employees.type IN (__[POSTCOMPILE_type_1])",
)
def test_outer_join_no_onclause_alias(self):
"employees_1.company_id AS employees_1_company_id "
"FROM companies LEFT OUTER JOIN employees AS employees_1 ON "
"companies.company_id = employees_1.company_id "
- "AND employees_1.type IN ([POSTCOMPILE_type_1])",
+ "AND employees_1.type IN (__[POSTCOMPILE_type_1])",
)
def test_correlated_column_select(self):
"(SELECT count(employees.employee_id) AS count_1 "
"FROM employees WHERE employees.company_id = "
"companies.company_id "
- "AND employees.type IN ([POSTCOMPILE_type_1])) AS anon_1 "
+ "AND employees.type IN (__[POSTCOMPILE_type_1])) AS anon_1 "
"FROM companies",
)
"ON companies.company_id = employees.company_id "
"JOIN employees "
"ON companies.company_id = employees.company_id "
- "AND employees.type IN ([POSTCOMPILE_type_1]) "
- "WHERE employees.type IN ([POSTCOMPILE_type_2])",
+ "AND employees.type IN (__[POSTCOMPILE_type_1]) "
+ "WHERE employees.type IN (__[POSTCOMPILE_type_2])",
)
def test_relationship_to_subclass(self):
"child.name AS child_name "
"FROM parent LEFT OUTER JOIN (m2m AS m2m_1 "
"JOIN child ON child.id = m2m_1.child_id "
- "AND child.discriminator IN ([POSTCOMPILE_discriminator_1])) "
+ "AND child.discriminator IN (__[POSTCOMPILE_discriminator_1])) "
"ON parent.id = m2m_1.parent_id",
)
"FROM parent LEFT OUTER JOIN "
"(m2m AS m2m_1 JOIN child AS child_1 "
"ON child_1.id = m2m_1.child_id AND child_1.discriminator "
- "IN ([POSTCOMPILE_discriminator_1])) "
+ "IN (__[POSTCOMPILE_discriminator_1])) "
"ON parent.id = m2m_1.parent_id",
)
"engineer.manager_id AS engineer_manager_id "
"FROM employee JOIN engineer ON employee.id = engineer.id) "
"AS anon_1 "
- "WHERE anon_1.employee_type IN ([POSTCOMPILE_type_1])",
+ "WHERE anon_1.employee_type IN (__[POSTCOMPILE_type_1])",
)
def test_query_wpoly_single_inh_subclass(self):
"engineer.engineer_info AS engineer_engineer_info, "
"engineer.manager_id AS engineer_manager_id "
"FROM employee JOIN engineer ON employee.id = engineer.id) "
- "AS anon_1 WHERE anon_1.employee_type IN ([POSTCOMPILE_type_1])",
+ "AS anon_1 WHERE anon_1.employee_type IN (__[POSTCOMPILE_type_1])",
)
@testing.combinations((True,), (False,), argnames="autoalias")
"JOIN (employee AS employee_1 JOIN engineer AS engineer_1 "
"ON employee_1.id = engineer_1.id) "
"ON engineer_1.manager_id = manager.id "
- "WHERE employee.type IN ([POSTCOMPILE_type_1])",
+ "WHERE employee.type IN (__[POSTCOMPILE_type_1])",
)
def test_single_inh_subclass_join_wpoly_joined_inh_subclass(self):
"FROM employee "
"JOIN engineer ON employee.id = engineer.id) AS anon_1 "
"ON anon_1.manager_id = manager.id "
- "WHERE employee.type IN ([POSTCOMPILE_type_1])",
+ "WHERE employee.type IN (__[POSTCOMPILE_type_1])",
)
@testing.combinations((True,), (False,), argnames="autoalias")
"JOIN (employee AS employee_1 JOIN manager AS manager_1 "
"ON employee_1.id = manager_1.id) "
"ON engineer.manager_id = manager_1.id "
- "AND employee_1.type IN ([POSTCOMPILE_type_1])",
+ "AND employee_1.type IN (__[POSTCOMPILE_type_1])",
)
"SELECT a_1.id AS a_1_id, b.id AS b_id FROM a AS a_1 "
"JOIN (b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) "
"ON a_1.b_id = b.id WHERE a_1.id "
- "IN ([POSTCOMPILE_primary_keys])",
+ "IN (__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [1]}],
),
)
expected = [
(
"SELECT users.id AS users_id, users.name AS users_name "
- "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])",
+ "FROM users WHERE users.id IN (__[POSTCOMPILE_id_1])",
{"id_1": [7, 8]},
),
(
lambda users: users.select().where(users.c.id.in_([7, 8])),
"SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name "
"FROM (SELECT users.id AS id, users.name AS name "
- "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "FROM users WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"WHERE anon_1.name = :name_1",
),
(
"SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name "
"AS anon_1_users_name FROM (SELECT users.id AS users_id, "
"users.name AS users_name FROM users "
- "WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"WHERE anon_1.users_name = :name_1",
),
(
lambda User, sess: sess.query(User).where(User.id.in_([7, 8])),
"SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name "
"FROM (SELECT users.id AS id, users.name AS name "
- "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "FROM users WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"WHERE anon_1.name = :name_1",
),
)
"users_1.name AS users_1_name "
"FROM users AS users_1, ("
"SELECT users.id AS id, users.name AS name FROM users "
- "WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"WHERE users_1.id > anon_1.id",
check_post_param={"id_1": [7, 8]},
)
"SELECT users_1.id AS users_1_id, "
"users_1.name AS users_1_name "
"FROM (SELECT users.id AS id, users.name AS name "
- "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "FROM users WHERE users.id IN "
+ "(__[POSTCOMPILE_id_1])) AS anon_1 "
"JOIN users AS users_1 ON users_1.id > anon_1.id",
check_post_param={"id_1": [7, 8]},
)
"SELECT users_1.id AS users_1_id, "
"users_1.name AS users_1_name "
"FROM (SELECT users.id AS id, users.name AS name FROM "
- "users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "users WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"JOIN users AS users_1 ON users_1.id > anon_1.id",
check_post_param={"id_1": [7, 8]},
)
"FROM "
"(SELECT users.id AS id, users.name AS name "
"FROM users WHERE users.id "
- "IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"JOIN users AS users_1 ON users_1.id > anon_1.id",
check_post_param={"id_1": [7, 8]},
)
sess.query(User).join(sel, User.id > sel.c.id),
"SELECT users.id AS users_id, users.name AS users_name FROM "
"users JOIN (SELECT users.id AS id, users.name AS name FROM users "
- "WHERE users.id IN ([POSTCOMPILE_id_1])) "
+ "WHERE users.id IN (__[POSTCOMPILE_id_1])) "
"AS anon_1 ON users.id > anon_1.id",
)
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM users AS users_1, ("
"SELECT users.id AS id, users.name AS name FROM users "
- "WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"WHERE users_1.id > anon_1.id",
check_post_param={"id_1": [7, 8]},
)
sess.query(ualias).select_from(ua).join(ualias, ualias.id > ua.id),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM (SELECT users.id AS id, users.name AS name "
- "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "FROM users WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"JOIN users AS users_1 ON users_1.id > anon_1.id",
check_post_param={"id_1": [7, 8]},
)
sess.query(ualias).select_from(ua).join(ualias, ualias.id > ua.id),
"SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
"FROM (SELECT users.id AS id, users.name AS name FROM "
- "users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "users WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"JOIN users AS users_1 ON users_1.id > anon_1.id",
check_post_param={"id_1": [7, 8]},
)
sess.query(salias).join(ualias, ualias.id > salias.id),
"SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name FROM "
"(SELECT users.id AS id, users.name AS name "
- "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
+ "FROM users WHERE users.id IN (__[POSTCOMPILE_id_1])) AS anon_1 "
"JOIN users AS users_1 ON users_1.id > anon_1.id",
check_post_param={"id_1": [7, 8]},
)
"FROM (SELECT anon_2.users_id AS users_id, "
"anon_2.users_name AS users_name FROM "
"(SELECT users.id AS users_id, users.name AS users_name "
- "FROM users) anon_2 WHERE ROWNUM <= [POSTCOMPILE_param_1]) anon_1 "
+ "FROM users) anon_2 WHERE ROWNUM <= "
+ "__[POSTCOMPILE_param_1]) anon_1 "
"LEFT OUTER JOIN addresses addresses_1 "
"ON anon_1.users_id = addresses_1.user_id FOR UPDATE",
dialect="oracle",
"c.id AS c_id, c.type AS c_type, c.b_id AS c_b_id, a.id AS a_id, "
"a.type AS a_type "
"FROM a LEFT OUTER JOIN b ON "
- "a.id = b.a_id AND b.type IN ([POSTCOMPILE_type_1]) "
+ "a.id = b.a_id AND b.type IN (__[POSTCOMPILE_type_1]) "
"LEFT OUTER JOIN c ON "
- "b.id = c.b_id AND c.type IN ([POSTCOMPILE_type_2]) "
- "WHERE a.type IN ([POSTCOMPILE_type_3])"
+ "b.id = c.b_id AND c.type IN (__[POSTCOMPILE_type_2]) "
+ "WHERE a.type IN (__[POSTCOMPILE_type_3])"
)
_query2 = (
"ccc.id AS ccc_id, ccc.type AS ccc_type, ccc.b_id AS ccc_b_id, "
"aaa.id AS aaa_id, aaa.type AS aaa_type "
"FROM a AS aaa LEFT OUTER JOIN b AS bbb "
- "ON aaa.id = bbb.a_id AND bbb.type IN ([POSTCOMPILE_type_1]) "
+ "ON aaa.id = bbb.a_id AND bbb.type IN (__[POSTCOMPILE_type_1]) "
"LEFT OUTER JOIN c AS ccc ON "
- "bbb.id = ccc.b_id AND ccc.type IN ([POSTCOMPILE_type_2]) "
- "WHERE aaa.type IN ([POSTCOMPILE_type_3])"
+ "bbb.id = ccc.b_id AND ccc.type IN (__[POSTCOMPILE_type_2]) "
+ "WHERE aaa.type IN (__[POSTCOMPILE_type_3])"
)
_query3 = (
"c.id AS c_id, c.type AS c_type, c.b_id AS c_b_id, "
"aaa.id AS aaa_id, aaa.type AS aaa_type "
"FROM a AS aaa LEFT OUTER JOIN b AS bbb "
- "ON aaa.id = bbb.a_id AND bbb.type IN ([POSTCOMPILE_type_1]) "
+ "ON aaa.id = bbb.a_id AND bbb.type IN (__[POSTCOMPILE_type_1]) "
"LEFT OUTER JOIN c ON "
- "bbb.id = c.b_id AND c.type IN ([POSTCOMPILE_type_2]) "
- "WHERE aaa.type IN ([POSTCOMPILE_type_3])"
+ "bbb.id = c.b_id AND c.type IN (__[POSTCOMPILE_type_2]) "
+ "WHERE aaa.type IN (__[POSTCOMPILE_type_3])"
)
def _test(self, join_of_type, of_type_for_c1, aliased_):
def test_in(self):
User = self.classes.User
- self._test(User.id.in_(["a", "b"]), "users.id IN ([POSTCOMPILE_id_1])")
+ self._test(
+ User.id.in_(["a", "b"]), "users.id IN (__[POSTCOMPILE_id_1])"
+ )
def test_in_on_relationship_not_supported(self):
User, Address = self.classes.User, self.classes.Address
"SELECT addresses.user_id AS addresses_user_id, addresses.id "
"AS addresses_id, addresses.email_address "
"AS addresses_email_address FROM addresses "
- "WHERE addresses.user_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE addresses.user_id IN (__[POSTCOMPILE_primary_keys]) "
"AND addresses.email_address != :email_address_1 "
"ORDER BY addresses.id",
[{"primary_keys": [7, 8, 9, 10], "email_address_1": "name"}],
"SELECT addresses.user_id AS addresses_user_id, "
"addresses.id AS addresses_id, addresses.email_address "
"AS addresses_email_address FROM addresses "
- "WHERE addresses.user_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE addresses.user_id IN "
+ "(__[POSTCOMPILE_primary_keys]) "
"AND addresses.email_address != :email_address_1 "
"ORDER BY addresses.id",
[
"ON items_1.id = order_items_1.item_id "
"AND items_1.description = :description_1) "
"ON orders.id = order_items_1.order_id "
- "WHERE orders.user_id IN ([POSTCOMPILE_primary_keys]) "
+ "WHERE orders.user_id IN (__[POSTCOMPILE_primary_keys]) "
"AND orders.description = :description_2 "
"ORDER BY orders.id, items_1.id",
[
"(SELECT a.id AS aid, b.id AS id FROM a JOIN b ON a.b_ids "
"LIKE :id_1 || b.id || :param_1) AS anon_1 "
"ON a_1.id = anon_1.aid JOIN b ON b.id = anon_1.id "
- "WHERE a_1.id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE a_1.id IN (__[POSTCOMPILE_primary_keys])",
params=[{"id_1": "%", "param_1": "%", "primary_keys": [2]}],
),
)
"paperwork.paperwork_id AS paperwork_paperwork_id, "
"paperwork.description AS paperwork_description "
"FROM paperwork WHERE paperwork.person_id "
- "IN ([POSTCOMPILE_primary_keys]) "
+ "IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY paperwork.paperwork_id",
[{"primary_keys": [1]}],
),
"paperwork.paperwork_id AS paperwork_paperwork_id, "
"paperwork.description AS paperwork_description "
"FROM paperwork WHERE paperwork.person_id "
- "IN ([POSTCOMPILE_primary_keys]) "
+ "IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY paperwork.paperwork_id",
[{"primary_keys": [1]}],
),
"paperwork.paperwork_id AS paperwork_paperwork_id, "
"paperwork.description AS paperwork_description "
"FROM paperwork WHERE paperwork.person_id "
- "IN ([POSTCOMPILE_primary_keys]) "
+ "IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY paperwork.paperwork_id",
[{"primary_keys": [1]}],
),
"paperwork.paperwork_id AS paperwork_paperwork_id, "
"paperwork.description AS paperwork_description "
"FROM paperwork WHERE paperwork.person_id "
- "IN ([POSTCOMPILE_primary_keys]) "
+ "IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY paperwork.paperwork_id",
[{"primary_keys": [1]}],
),
"paperwork.paperwork_id AS paperwork_paperwork_id, "
"paperwork.description AS paperwork_description "
"FROM paperwork WHERE paperwork.person_id "
- "IN ([POSTCOMPILE_primary_keys]) "
+ "IN (__[POSTCOMPILE_primary_keys]) "
"ORDER BY paperwork.paperwork_id",
[{"primary_keys": [1]}],
),
CompiledSQL(
"SELECT b.a_id1 AS b_a_id1, b.a_id2 AS b_a_id2, b.id AS b_id "
"FROM b WHERE (b.a_id1, b.a_id2) IN "
- "([POSTCOMPILE_primary_keys]) ORDER BY b.id",
+ "(__[POSTCOMPILE_primary_keys]) ORDER BY b.id",
[{"primary_keys": [(i, i + 2) for i in range(1, 20)]}],
),
)
),
CompiledSQL(
"SELECT a.id1 AS a_id1, a.id2 AS a_id2 FROM a "
- "WHERE (a.id1, a.id2) IN ([POSTCOMPILE_primary_keys])",
+ "WHERE (a.id1, a.id2) IN (__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [(i, i + 2) for i in range(1, 20)]}],
),
)
CompiledSQL(
"SELECT b.a_id AS b_a_id, b.id AS b_id "
"FROM b WHERE b.a_id IN "
- "([POSTCOMPILE_primary_keys]) ORDER BY b.id",
+ "(__[POSTCOMPILE_primary_keys]) ORDER BY b.id",
{"primary_keys": list(range(1, 48))},
),
CompiledSQL(
"SELECT b.a_id AS b_a_id, b.id AS b_id "
"FROM b WHERE b.a_id IN "
- "([POSTCOMPILE_primary_keys]) ORDER BY b.id",
+ "(__[POSTCOMPILE_primary_keys]) ORDER BY b.id",
{"primary_keys": list(range(48, 95))},
),
CompiledSQL(
"SELECT b.a_id AS b_a_id, b.id AS b_id "
"FROM b WHERE b.a_id IN "
- "([POSTCOMPILE_primary_keys]) ORDER BY b.id",
+ "(__[POSTCOMPILE_primary_keys]) ORDER BY b.id",
{"primary_keys": list(range(95, 101))},
),
)
# chunk size is 47. so first chunk are a 1->47...
CompiledSQL(
"SELECT a.id AS a_id FROM a WHERE a.id IN "
- "([POSTCOMPILE_primary_keys])",
+ "(__[POSTCOMPILE_primary_keys])",
{"primary_keys": list(range(1, 48))},
),
# second chunk is a 48-94
CompiledSQL(
"SELECT a.id AS a_id FROM a WHERE a.id IN "
- "([POSTCOMPILE_primary_keys])",
+ "(__[POSTCOMPILE_primary_keys])",
{"primary_keys": list(range(48, 95))},
),
# third and final chunk 95-100.
CompiledSQL(
"SELECT a.id AS a_id FROM a WHERE a.id IN "
- "([POSTCOMPILE_primary_keys])",
+ "(__[POSTCOMPILE_primary_keys])",
{"primary_keys": list(range(95, 101))},
),
)
"SELECT foo_1.id AS foo_1_id, "
"foo_1.type AS foo_1_type, foo_1.foo_id AS foo_1_foo_id "
"FROM foo AS foo_1 "
- "WHERE foo_1.id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE foo_1.id IN (__[POSTCOMPILE_primary_keys])",
{"primary_keys": [3]},
),
CompiledSQL(
"SELECT foo.id AS foo_id_1, foo.type AS foo_type, "
"foo.foo_id AS foo_foo_id FROM foo "
- "WHERE foo.id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE foo.id IN (__[POSTCOMPILE_primary_keys])",
{"primary_keys": [1]},
),
)
q.all,
CompiledSQL(
'SELECT "user".id AS user_id, "user".type AS user_type '
- 'FROM "user" WHERE "user".type IN ([POSTCOMPILE_type_1])',
+ 'FROM "user" WHERE "user".type IN (__[POSTCOMPILE_type_1])',
{"type_1": ["employer"]},
),
CompiledSQL(
"SELECT role.user_id AS role_user_id, role.id AS role_id "
"FROM role WHERE role.user_id "
- "IN ([POSTCOMPILE_primary_keys])",
+ "IN (__[POSTCOMPILE_primary_keys])",
{"primary_keys": [1]},
),
)
q.all,
CompiledSQL(
"SELECT a.id AS a_id, a.b_id AS a_b_id, a.q AS a_q "
- "FROM a WHERE a.id IN ([POSTCOMPILE_id_1]) ORDER BY a.id",
+ "FROM a WHERE a.id IN (__[POSTCOMPILE_id_1]) ORDER BY a.id",
[{"id_1": [1, 3]}],
),
CompiledSQL(
"SELECT b.id AS b_id, b.x AS b_x, b.y AS b_y "
- "FROM b WHERE b.id IN ([POSTCOMPILE_primary_keys])",
+ "FROM b WHERE b.id IN (__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [1, 2]}],
),
)
q.all,
CompiledSQL(
"SELECT a.id AS a_id, a.q AS a_q "
- "FROM a WHERE a.id IN ([POSTCOMPILE_id_1]) ORDER BY a.id",
+ "FROM a WHERE a.id IN (__[POSTCOMPILE_id_1]) ORDER BY a.id",
[{"id_1": [1, 3]}],
),
# in the very unlikely case that the the FK col on parent is
"SELECT a_1.id AS a_1_id, b.id AS b_id, b.x AS b_x, "
"b.y AS b_y "
"FROM a AS a_1 JOIN b ON b.id = a_1.b_id "
- "WHERE a_1.id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE a_1.id IN (__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [1, 3]}],
),
)
),
CompiledSQL(
"SELECT b.id AS b_id, b.x AS b_x, b.y AS b_y "
- "FROM b WHERE b.id IN ([POSTCOMPILE_primary_keys])",
+ "FROM b WHERE b.id IN (__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [1, 2]}],
),
)
CompiledSQL(
"SELECT a_1.id AS a_1_id, b.id AS b_id, b.x AS b_x, "
"b.y AS b_y FROM a AS a_1 JOIN b ON b.id = a_1.b_id "
- "WHERE a_1.id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE a_1.id IN (__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [1, 2, 3, 4, 5]}],
),
)
"SELECT a_1.id AS a_1_id, b.id AS b_id, b.x AS b_x, "
"b.y AS b_y "
"FROM a AS a_1 JOIN b ON b.id = a_1.b_id "
- "WHERE a_1.id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE a_1.id IN (__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [1, 2, 3, 4, 5]}],
),
)
CompiledSQL(
"SELECT child_a.parent_id AS child_a_parent_id, "
"child_a.id AS child_a_id FROM child_a "
- "WHERE child_a.parent_id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE child_a.parent_id IN "
+ "(__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [1]}],
),
CompiledSQL(
"SELECT child_b.parent_id AS child_b_parent_id, "
"child_b.id AS child_b_id FROM child_b "
- "WHERE child_b.parent_id IN ([POSTCOMPILE_primary_keys])",
+ "WHERE child_b.parent_id IN "
+ "(__[POSTCOMPILE_primary_keys])",
[{"primary_keys": [2]}],
),
),
)
self.assert_compile(
expr,
- "(mytable.myid, mytable.name) IN " "([POSTCOMPILE_param_1])",
+ "(mytable.myid, mytable.name) IN " "(__[POSTCOMPILE_param_1])",
checkparams={"param_1": [(1, "foo"), (5, "bar")]},
check_post_param={"param_1": [(1, "foo"), (5, "bar")]},
check_literal_execute={},
dialect.tuple_in_values = True
self.assert_compile(
tuple_(table1.c.myid, table1.c.name).in_([(1, "foo"), (5, "bar")]),
- "(mytable.myid, mytable.name) IN " "([POSTCOMPILE_param_1])",
+ "(mytable.myid, mytable.name) IN " "(__[POSTCOMPILE_param_1])",
dialect=dialect,
checkparams={"param_1": [(1, "foo"), (5, "bar")]},
check_post_param={"param_1": [(1, "foo"), (5, "bar")]},
tuple_(table1.c.myid, table1.c.name).in_(
bindparam("foo", expanding=True)
),
- "(mytable.myid, mytable.name) IN ([POSTCOMPILE_foo])",
+ "(mytable.myid, mytable.name) IN (__[POSTCOMPILE_foo])",
)
dialect = default.DefaultDialect()
tuple_(table1.c.myid, table1.c.name).in_(
bindparam("foo", expanding=True)
),
- "(mytable.myid, mytable.name) IN ([POSTCOMPILE_foo])",
+ "(mytable.myid, mytable.name) IN (__[POSTCOMPILE_foo])",
dialect=dialect,
)
self.assert_compile(
table1.c.myid.in_(bindparam("foo", expanding=True)),
- "mytable.myid IN ([POSTCOMPILE_foo])",
+ "mytable.myid IN (__[POSTCOMPILE_foo])",
)
def test_limit_offset_select_literal_binds(self):
(
"one",
select(literal("someliteral")),
- "SELECT [POSTCOMPILE_param_1] AS anon_1",
+ "SELECT __[POSTCOMPILE_param_1] AS anon_1",
dict(
check_literal_execute={"param_1": "someliteral"},
check_post_param={},
(
"two",
select(table1.c.myid + 3),
- "SELECT mytable.myid + [POSTCOMPILE_myid_1] "
+ "SELECT mytable.myid + __[POSTCOMPILE_myid_1] "
"AS anon_1 FROM mytable",
dict(check_literal_execute={"myid_1": 3}, check_post_param={}),
),
(
"three",
select(table1.c.myid.in_([4, 5, 6])),
- "SELECT mytable.myid IN ([POSTCOMPILE_myid_1]) "
+ "SELECT mytable.myid IN (__[POSTCOMPILE_myid_1]) "
"AS anon_1 FROM mytable",
dict(
check_literal_execute={"myid_1": [4, 5, 6]},
(
"four",
select(func.mod(table1.c.myid, 5)),
- "SELECT mod(mytable.myid, [POSTCOMPILE_mod_2]) "
+ "SELECT mod(mytable.myid, __[POSTCOMPILE_mod_2]) "
"AS mod_1 FROM mytable",
dict(check_literal_execute={"mod_2": 5}, check_post_param={}),
),
(
"five",
select(literal("foo").in_([])),
- "SELECT [POSTCOMPILE_param_1] IN ([POSTCOMPILE_param_2]) "
+ "SELECT __[POSTCOMPILE_param_1] IN (__[POSTCOMPILE_param_2]) "
"AS anon_1",
dict(
check_literal_execute={"param_1": "foo", "param_2": []},
(
"six",
select(literal(util.b("foo"))),
- "SELECT [POSTCOMPILE_param_1] AS anon_1",
+ "SELECT __[POSTCOMPILE_param_1] AS anon_1",
dict(
check_literal_execute={"param_1": util.b("foo")},
check_post_param={},
(
"seven",
select(table1.c.myid == bindparam("foo", callable_=lambda: 5)),
- "SELECT mytable.myid = [POSTCOMPILE_foo] AS anon_1 FROM mytable",
+ "SELECT mytable.myid = __[POSTCOMPILE_foo] AS anon_1 FROM mytable",
dict(check_literal_execute={"foo": 5}, check_post_param={}),
),
argnames="stmt, expected, kw",
table1.c.myid == bindparam("foo", 5, literal_execute=True)
),
"SELECT mytable.myid FROM mytable "
- "WHERE mytable.myid = [POSTCOMPILE_foo]",
+ "WHERE mytable.myid = __[POSTCOMPILE_foo]",
)
def test_render_literal_execute_parameter_literal_binds(self):
table1.c.myid.in_(bindparam("foo", expanding=True))
),
"SELECT mytable.myid FROM mytable "
- "WHERE mytable.myid IN ([POSTCOMPILE_foo])",
+ "WHERE mytable.myid IN (__[POSTCOMPILE_foo])",
)
def test_render_expanding_parameter_literal_binds(self):
self.assert_compile(
schema.CreateTable(t1),
- "CREATE TABLE [SCHEMA__none].t1 (q INTEGER)",
+ "CREATE TABLE __[SCHEMA__none].t1 (q INTEGER)",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
self.assert_compile(
schema.CreateTable(t2),
- "CREATE TABLE [SCHEMA_foo].t2 (q INTEGER)",
+ "CREATE TABLE __[SCHEMA_foo].t2 (q INTEGER)",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
self.assert_compile(
schema.CreateTable(t3),
- "CREATE TABLE [SCHEMA_bar].t3 (q INTEGER)",
+ "CREATE TABLE __[SCHEMA_bar].t3 (q INTEGER)",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
self.assert_compile(
schema.CreateTable(t1),
- "CREATE TABLE [SCHEMA__none].t1 (q INTEGER)",
+ "CREATE TABLE __[SCHEMA__none].t1 (q INTEGER)",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
self.assert_compile(
schema.CreateTable(t2),
- "CREATE TABLE [SCHEMA_foo % ^ #].t2 (q INTEGER)",
+ "CREATE TABLE __[SCHEMA_foo % ^ #].t2 (q INTEGER)",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
self.assert_compile(
schema.CreateTable(t3),
- "CREATE TABLE [SCHEMA_bar {}].t3 (q INTEGER)",
+ "CREATE TABLE __[SCHEMA_bar {}].t3 (q INTEGER)",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
self.assert_compile(
schema.CreateSequence(s1),
- "CREATE SEQUENCE [SCHEMA__none].s1 START WITH 1",
+ "CREATE SEQUENCE __[SCHEMA__none].s1 START WITH 1",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
s1.next_value(),
- "<next sequence value: [SCHEMA__none].s1>",
+ "<next sequence value: __[SCHEMA__none].s1>",
schema_translate_map=schema_translate_map,
dialect="default_enhanced",
)
self.assert_compile(
schema.CreateSequence(s2),
- "CREATE SEQUENCE [SCHEMA_foo].s2 START WITH 1",
+ "CREATE SEQUENCE __[SCHEMA_foo].s2 START WITH 1",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
s2.next_value(),
- "<next sequence value: [SCHEMA_foo].s2>",
+ "<next sequence value: __[SCHEMA_foo].s2>",
schema_translate_map=schema_translate_map,
dialect="default_enhanced",
)
self.assert_compile(
schema.CreateSequence(s3),
- "CREATE SEQUENCE [SCHEMA_bar].s3 START WITH 1",
+ "CREATE SEQUENCE __[SCHEMA_bar].s3 START WITH 1",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
s3.next_value(),
- "<next sequence value: [SCHEMA_bar].s3>",
+ "<next sequence value: __[SCHEMA_bar].s3>",
schema_translate_map=schema_translate_map,
dialect="default_enhanced",
)
self.assert_compile(
schema.CreateTable(t1),
- "CREATE TABLE [SCHEMA__none].t1 "
- "(id INTEGER DEFAULT <next sequence value: [SCHEMA__none].s1> "
+ "CREATE TABLE __[SCHEMA__none].t1 "
+ "(id INTEGER DEFAULT <next sequence value: __[SCHEMA__none].s1> "
"NOT NULL, PRIMARY KEY (id))",
schema_translate_map=schema_translate_map,
dialect="default_enhanced",
)
self.assert_compile(
schema.CreateTable(t2),
- "CREATE TABLE [SCHEMA__none].t2 "
- "(id INTEGER DEFAULT <next sequence value: [SCHEMA_foo].s2> "
+ "CREATE TABLE __[SCHEMA__none].t2 "
+ "(id INTEGER DEFAULT <next sequence value: __[SCHEMA_foo].s2> "
"NOT NULL, PRIMARY KEY (id))",
schema_translate_map=schema_translate_map,
dialect="default_enhanced",
)
self.assert_compile(
schema.CreateTable(t3),
- "CREATE TABLE [SCHEMA__none].t3 "
- "(id INTEGER DEFAULT <next sequence value: [SCHEMA_bar].s3> "
+ "CREATE TABLE __[SCHEMA__none].t3 "
+ "(id INTEGER DEFAULT <next sequence value: __[SCHEMA_bar].s3> "
"NOT NULL, PRIMARY KEY (id))",
schema_translate_map=schema_translate_map,
dialect="default_enhanced",
self.assert_compile(
stmt,
- "SELECT [SCHEMA__none].myothertable.otherid, "
- "[SCHEMA__none].myothertable.othername, "
+ "SELECT __[SCHEMA__none].myothertable.otherid, "
+ "__[SCHEMA__none].myothertable.othername, "
"mytable_1.myid, mytable_1.name, mytable_1.description "
- "FROM [SCHEMA__none].myothertable JOIN "
- "[SCHEMA__none].mytable AS mytable_1 "
- "ON [SCHEMA__none].myothertable.otherid = mytable_1.myid "
+ "FROM __[SCHEMA__none].myothertable JOIN "
+ "__[SCHEMA__none].mytable AS mytable_1 "
+ "ON __[SCHEMA__none].myothertable.otherid = mytable_1.myid "
"WHERE mytable_1.name = :name_1",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
column("x").notin_(["foo", "bar"]),
- "(x NOT IN ([POSTCOMPILE_x_1]))",
+ "(x NOT IN (__[POSTCOMPILE_x_1]))",
)
def test_issue_5429_operators(self):
and the compiler postcompile reg is::
- re.sub(r"\[POSTCOMPILE_(\S+)\]", process_expanding, self.string)
+ re.sub(r"\__[POSTCOMPILE_(\S+)\]", process_expanding, self.string)
Interestingly, brackets in the name seems to work out.
stmt = and_(expr, expr2)
self.assert_compile(
- stmt, "x IN ([POSTCOMPILE_x_1]) AND x IN ([POSTCOMPILE_x_1])"
+ stmt, "x IN (__[POSTCOMPILE_x_1]) AND x IN (__[POSTCOMPILE_x_1])"
)
self.assert_compile(
stmt, "x IN (1, 2, 3) AND x IN (1, 2, 3)", literal_binds=True
asserter_.assert_(
CompiledSQL(
"SELECT users.id FROM users WHERE users.name "
- "IN ([POSTCOMPILE_val_1]) ORDER BY users.id",
+ "IN (__[POSTCOMPILE_val_1]) ORDER BY users.id",
params={"val_1": case},
)
)
def test_in_parameters_one(self):
expr1 = select(1).where(column("q").in_(["a", "b", "c"]))
- self.assert_compile(expr1, "SELECT 1 WHERE q IN ([POSTCOMPILE_q_1])")
+ self.assert_compile(expr1, "SELECT 1 WHERE q IN (__[POSTCOMPILE_q_1])")
self.assert_compile(
expr1,
def test_in_parameters_two(self):
expr2 = select(1).where(lambda: column("q").in_(["a", "b", "c"]))
- self.assert_compile(expr2, "SELECT 1 WHERE q IN ([POSTCOMPILE_q_1])")
+ self.assert_compile(expr2, "SELECT 1 WHERE q IN (__[POSTCOMPILE_q_1])")
self.assert_compile(
expr2,
"SELECT 1 WHERE q IN (:q_1_1, :q_1_2, :q_1_3)",
expr3 = lambdas.lambda_stmt(
lambda: select(1).where(column("q").in_(["a", "b", "c"]))
)
- self.assert_compile(expr3, "SELECT 1 WHERE q IN ([POSTCOMPILE_q_1])")
+ self.assert_compile(expr3, "SELECT 1 WHERE q IN (__[POSTCOMPILE_q_1])")
self.assert_compile(
expr3,
"SELECT 1 WHERE q IN (:q_1_1, :q_1_2, :q_1_3)",
expr4 = go(["a", "b", "c"])
self.assert_compile(
- expr4, "SELECT 1 WHERE q IN ([POSTCOMPILE_names_1])"
+ expr4, "SELECT 1 WHERE q IN (__[POSTCOMPILE_names_1])"
)
self.assert_compile(
expr4,
opts=lambdas.LambdaOptions(track_closure_variables=False),
)
- self.assert_compile(elem.expr, "t1.q IN ([POSTCOMPILE_vv_1])")
+ self.assert_compile(elem.expr, "t1.q IN (__[POSTCOMPILE_vv_1])")
assert_raises_message(
exc.InvalidRequestError,
self.assert_compile(
self.table2.select().where(5 + self.table2.c.field.in_([5, 6])),
"SELECT op.field FROM op WHERE :param_1 + "
- "(op.field IN ([POSTCOMPILE_field_1]))",
+ "(op.field IN (__[POSTCOMPILE_field_1]))",
)
def test_operator_precedence_6(self):
self.assert_compile(
self.table2.select().where((5 + self.table2.c.field).in_([5, 6])),
"SELECT op.field FROM op WHERE :field_1 + op.field "
- "IN ([POSTCOMPILE_param_1])",
+ "IN (__[POSTCOMPILE_param_1])",
)
def test_operator_precedence_7(self):
def test_in_1(self):
self.assert_compile(
self.table1.c.myid.in_(["a"]),
- "mytable.myid IN ([POSTCOMPILE_myid_1])",
+ "mytable.myid IN (__[POSTCOMPILE_myid_1])",
checkparams={"myid_1": ["a"]},
)
def test_in_2(self):
self.assert_compile(
~self.table1.c.myid.in_(["a"]),
- "(mytable.myid NOT IN ([POSTCOMPILE_myid_1]))",
+ "(mytable.myid NOT IN (__[POSTCOMPILE_myid_1]))",
checkparams={"myid_1": ["a"]},
)
def test_in_3(self):
self.assert_compile(
self.table1.c.myid.in_(["a", "b"]),
- "mytable.myid IN ([POSTCOMPILE_myid_1])",
+ "mytable.myid IN (__[POSTCOMPILE_myid_1])",
checkparams={"myid_1": ["a", "b"]},
)
def test_in_4(self):
self.assert_compile(
self.table1.c.myid.in_(iter(["a", "b"])),
- "mytable.myid IN ([POSTCOMPILE_myid_1])",
+ "mytable.myid IN (__[POSTCOMPILE_myid_1])",
checkparams={"myid_1": ["a", "b"]},
)
def test_in_19(self):
self.assert_compile(
self.table1.c.myid.in_([1, 2, 3]),
- "mytable.myid IN ([POSTCOMPILE_myid_1])",
+ "mytable.myid IN (__[POSTCOMPILE_myid_1])",
checkparams={"myid_1": [1, 2, 3]},
)
if is_in:
self.assert_compile(
expr,
- "(a, b, c) %s ([POSTCOMPILE_param_1])"
+ "(a, b, c) %s (__[POSTCOMPILE_param_1])"
% ("IN" if is_in else "NOT IN"),
checkparams={"param_1": [(3, "hi", b"there"), (4, "Q", b"P")]},
)
else:
self.assert_compile(
expr,
- "((a, b, c) NOT IN ([POSTCOMPILE_param_1]))",
+ "((a, b, c) NOT IN (__[POSTCOMPILE_param_1]))",
checkparams={"param_1": [(3, "hi", b"there"), (4, "Q", b"P")]},
)
self.assert_compile(
if is_in:
self.assert_compile(
expr,
- "(a, b, c) IN ([POSTCOMPILE_param_1])",
+ "(a, b, c) IN (__[POSTCOMPILE_param_1])",
checkparams={"param_1": []},
)
self.assert_compile(
else:
self.assert_compile(
expr,
- "((a, b, c) NOT IN ([POSTCOMPILE_param_1]))",
+ "((a, b, c) NOT IN (__[POSTCOMPILE_param_1]))",
checkparams={"param_1": []},
)
self.assert_compile(
if is_in:
self.assert_compile(
expr,
- "a IN ([POSTCOMPILE_a_1])",
+ "a IN (__[POSTCOMPILE_a_1])",
checkparams={"a_1": []},
)
self.assert_compile(
else:
self.assert_compile(
expr,
- "(a NOT IN ([POSTCOMPILE_a_1]))",
+ "(a NOT IN (__[POSTCOMPILE_a_1]))",
checkparams={"a_1": []},
)
self.assert_compile(
stmt = and_(expr1, expr2)
self.assert_compile(
- stmt, "a IN ([POSTCOMPILE_a_1]) AND (a NOT IN ([POSTCOMPILE_a_2]))"
+ stmt,
+ "a IN (__[POSTCOMPILE_a_1]) AND (a NOT IN (__[POSTCOMPILE_a_2]))",
)
self.assert_compile(
stmt, "a IN (5) AND (a NOT IN (5))", literal_binds=True
stmt = and_(expr1, expr2)
self.assert_compile(
- stmt, "a IN ([POSTCOMPILE_a_1]) AND (a NOT IN ([POSTCOMPILE_a_2]))"
+ stmt,
+ "a IN (__[POSTCOMPILE_a_1]) AND (a NOT IN (__[POSTCOMPILE_a_2]))",
)
self.assert_compile(
stmt,
s = {1, 2, 3}
self.assert_compile(
self.table1.c.myid.in_(s),
- "mytable.myid IN ([POSTCOMPILE_myid_1])",
+ "mytable.myid IN (__[POSTCOMPILE_myid_1])",
checkparams={"myid_1": list(s)},
)
seq = MySeq([1, 2, 3])
self.assert_compile(
self.table1.c.myid.in_(seq),
- "mytable.myid IN ([POSTCOMPILE_myid_1])",
+ "mytable.myid IN (__[POSTCOMPILE_myid_1])",
checkparams={"myid_1": [1, 2, 3]},
)
(table1.c.col1 == 5, "table1.col1 = :col1_1"),
(
table1.c.col1.in_([2, 3, 4]),
- "table1.col1 IN ([POSTCOMPILE_col1_1])",
+ "table1.col1 IN (__[POSTCOMPILE_col1_1])",
),
]:
eq_(str(expr), expected)
),
"SELECT test_table.x, lower(test_table.y) AS y FROM "
"test_table WHERE test_table.y IN "
- "([POSTCOMPILE_y_1~~lower(~~REPL~~)~~])",
+ "(__[POSTCOMPILE_y_1~~lower(~~REPL~~)~~])",
render_postcompile=False,
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
