PublicArchive/sqlalchemy
2
0
Fork 0
mirror of https://github.com/sqlalchemy/sqlalchemy.git synced 2026-05-16 05:37:16 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
3710382de1c285c8a613c71be165bd192da86dfd
sqlalchemy/test/sql/test_lambdas.py
T
Federico Caselli e860060866 Update select usage to use the new 1.4 format 
This change includes mainly that the bracketed use within
select() is moved to positional, and keyword arguments are
removed from calls to the select() function.  it does not
yet fully address other issues such as keyword arguments passed
to the table.select().

Additionally, allows False / None to both be considered
as "disable" for all of select.correlate(), select.correlate_except(),
query.correlate(), which establishes consistency with
passing of ``False`` for the legact select(correlate=False)
argument.

Change-Id: Ie6c6e6abfbd3d75d4c8de504c0cf0159e6999108
2020-09-08 17:13:48 -04:00

1127 lines
33 KiB
Python
Raw Blame History

from sqlalchemy import exc
from sqlalchemy import testing
from sqlalchemy.future import select as future_select
from sqlalchemy.schema import Column
from sqlalchemy.schema import ForeignKey
from sqlalchemy.schema import Table
from sqlalchemy.sql import and_
from sqlalchemy.sql import bindparam
from sqlalchemy.sql import coercions
from sqlalchemy.sql import column
from sqlalchemy.sql import join
from sqlalchemy.sql import lambda_stmt
from sqlalchemy.sql import lambdas
from sqlalchemy.sql import roles
from sqlalchemy.sql import select
from sqlalchemy.sql import table
from sqlalchemy.sql import util as sql_util
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import ne_
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.types import Integer
from sqlalchemy.types import String
class DeferredLambdaTest(
fixtures.TestBase, testing.AssertsExecutionResults, AssertsCompiledSQL
):
__dialect__ = "default"
def test_select_whereclause(self):
t1 = table("t1", column("q"), column("p"))
x = 10
y = 5
def go():
return select(t1).where(lambda: and_(t1.c.q == x, t1.c.p == y))
self.assert_compile(
go(), "SELECT t1.q, t1.p FROM t1 WHERE t1.q = :x_1 AND t1.p = :y_1"
)
self.assert_compile(
go(), "SELECT t1.q, t1.p FROM t1 WHERE t1.q = :x_1 AND t1.p = :y_1"
)
def test_global_tracking(self):
t1 = table("t1", column("q"), column("p"))
global global_x, global_y
global_x = 10
global_y = 17
def go():
return select(t1).where(
lambda: and_(t1.c.q == global_x, t1.c.p == global_y)
)
self.assert_compile(
go(),
"SELECT t1.q, t1.p FROM t1 WHERE t1.q = :global_x_1 "
"AND t1.p = :global_y_1",
checkparams={"global_x_1": 10, "global_y_1": 17},
)
global_y = 9
self.assert_compile(
go(),
"SELECT t1.q, t1.p FROM t1 WHERE t1.q = :global_x_1 "
"AND t1.p = :global_y_1",
checkparams={"global_x_1": 10, "global_y_1": 9},
)
def test_stale_checker_embedded(self):
def go(x):
stmt = select(lambda: x)
return stmt
c1 = column("x")
s1 = go(c1)
s2 = go(c1)
self.assert_compile(s1, "SELECT x")
self.assert_compile(s2, "SELECT x")
c1 = column("q")
s3 = go(c1)
self.assert_compile(s3, "SELECT q")
def test_stale_checker_statement(self):
def go(x):
stmt = lambdas.lambda_stmt(lambda: select(x))
return stmt
c1 = column("x")
s1 = go(c1)
s2 = go(c1)
self.assert_compile(s1, "SELECT x")
self.assert_compile(s2, "SELECT x")
c1 = column("q")
s3 = go(c1)
self.assert_compile(s3, "SELECT q")
def test_stale_checker_linked(self):
def go(x, y):
stmt = lambdas.lambda_stmt(lambda: select(x)) + (
lambda s: s.where(y > 5)
)
return stmt
c1 = oldc1 = column("x")
c2 = oldc2 = column("y")
s1 = go(c1, c2)
s2 = go(c1, c2)
self.assert_compile(s1, "SELECT x WHERE y > :y_1")
self.assert_compile(s2, "SELECT x WHERE y > :y_1")
c1 = column("q")
c2 = column("p")
s3 = go(c1, c2)
self.assert_compile(s3, "SELECT q WHERE p > :p_1")
s4 = go(c1, c2)
self.assert_compile(s4, "SELECT q WHERE p > :p_1")
s5 = go(oldc1, oldc2)
self.assert_compile(s5, "SELECT x WHERE y > :y_1")
def test_stmt_lambda_w_additional_hascachekey_variants(self):
def go(col_expr, q):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt += lambda stmt: stmt.where(col_expr == q)
return stmt
c1 = column("x")
c2 = column("y")
s1 = go(c1, 5)
s2 = go(c2, 10)
s3 = go(c1, 8)
s4 = go(c2, 12)
self.assert_compile(
s1, "SELECT x WHERE x = :q_1", checkparams={"q_1": 5}
)
self.assert_compile(
s2, "SELECT y WHERE y = :q_1", checkparams={"q_1": 10}
)
self.assert_compile(
s3, "SELECT x WHERE x = :q_1", checkparams={"q_1": 8}
)
self.assert_compile(
s4, "SELECT y WHERE y = :q_1", checkparams={"q_1": 12}
)
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
s3key = s3._generate_cache_key()
s4key = s4._generate_cache_key()
eq_(s1key[0], s3key[0])
eq_(s2key[0], s4key[0])
ne_(s1key[0], s2key[0])
def test_stmt_lambda_w_atonce_whereclause_values_notrack(self):
def go(col_expr, whereclause):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt = stmt.add_criteria(
lambda stmt: stmt.where(whereclause), enable_tracking=False
)
return stmt
c1 = column("x")
s1 = go(c1, c1 == 5)
s2 = go(c1, c1 == 10)
self.assert_compile(
s1, "SELECT x WHERE x = :x_1", checkparams={"x_1": 5}
)
# and as we see, this is wrong. Because whereclause
# is fixed for the lambda and we do not re-evaluate the closure
# for this value changing. this can't be passed unless
# enable_tracking=False.
self.assert_compile(
s2, "SELECT x WHERE x = :x_1", checkparams={"x_1": 5}
)
def test_stmt_lambda_w_atonce_whereclause_values(self):
c2 = column("y")
def go(col_expr, whereclause, x):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt = stmt.add_criteria(
lambda stmt: stmt.where(whereclause).order_by(c2 > x),
)
return stmt
c1 = column("x")
s1 = go(c1, c1 == 5, 9)
s2 = go(c1, c1 == 10, 15)
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
eq_([b.value for b in s1key.bindparams], [5, 9])
eq_([b.value for b in s2key.bindparams], [10, 15])
self.assert_compile(
s1,
"SELECT x WHERE x = :x_1 ORDER BY y > :x_2",
checkparams={"x_1": 5, "x_2": 9},
)
self.assert_compile(
s2,
"SELECT x WHERE x = :x_1 ORDER BY y > :x_2",
checkparams={"x_1": 10, "x_2": 15},
)
def test_stmt_lambda_plain_customtrack(self):
c2 = column("y")
def go(col_expr, whereclause, p):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt = stmt.add_criteria(lambda stmt: stmt.where(whereclause))
stmt = stmt.add_criteria(
lambda stmt: stmt.order_by(col_expr), track_on=(col_expr,)
)
stmt = stmt.add_criteria(lambda stmt: stmt.where(col_expr == p))
return stmt
c1 = column("x")
c2 = column("y")
s1 = go(c1, c1 == 5, 9)
s2 = go(c1, c1 == 10, 15)
s3 = go(c2, c2 == 18, 12)
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
s3key = s3._generate_cache_key()
eq_([b.value for b in s1key.bindparams], [5, 9])
eq_([b.value for b in s2key.bindparams], [10, 15])
eq_([b.value for b in s3key.bindparams], [18, 12])
self.assert_compile(
s1,
"SELECT x WHERE x = :x_1 AND x = :p_1 ORDER BY x",
checkparams={"x_1": 5, "p_1": 9},
)
self.assert_compile(
s2,
"SELECT x WHERE x = :x_1 AND x = :p_1 ORDER BY x",
checkparams={"x_1": 10, "p_1": 15},
)
self.assert_compile(
s3,
"SELECT y WHERE y = :y_1 AND y = :p_1 ORDER BY y",
checkparams={"y_1": 18, "p_1": 12},
)
def test_stmt_lambda_w_atonce_whereclause_customtrack_binds(self):
c2 = column("y")
# this pattern is *completely unnecessary*, and I would prefer
# if we can detect this and just raise, because when it is not done
# correctly, it is *extremely* difficult to catch it failing.
# however I also can't come up with a reliable way to catch it.
# so we will keep the use of "track_on" to be internal.
def go(col_expr, whereclause, p):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt = stmt.add_criteria(
lambda stmt: stmt.where(whereclause).order_by(col_expr > p),
track_on=(whereclause, whereclause.right.value),
)
return stmt
c1 = column("x")
c2 = column("y")
s1 = go(c1, c1 == 5, 9)
s2 = go(c1, c1 == 10, 15)
s3 = go(c2, c2 == 18, 12)
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
s3key = s3._generate_cache_key()
eq_([b.value for b in s1key.bindparams], [5, 9])
eq_([b.value for b in s2key.bindparams], [10, 15])
eq_([b.value for b in s3key.bindparams], [18, 12])
self.assert_compile(
s1,
"SELECT x WHERE x = :x_1 ORDER BY x > :p_1",
checkparams={"x_1": 5, "p_1": 9},
)
self.assert_compile(
s2,
"SELECT x WHERE x = :x_1 ORDER BY x > :p_1",
checkparams={"x_1": 10, "p_1": 15},
)
self.assert_compile(
s3,
"SELECT y WHERE y = :y_1 ORDER BY y > :p_1",
checkparams={"y_1": 18, "p_1": 12},
)
def test_stmt_lambda_track_closure_binds_one(self):
def go(col_expr, whereclause):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt += lambda stmt: stmt.where(whereclause)
return stmt
c1 = column("x")
s1 = go(c1, c1 == 5)
s2 = go(c1, c1 == 10)
self.assert_compile(
s1, "SELECT x WHERE x = :x_1", checkparams={"x_1": 5}
)
self.assert_compile(
s2, "SELECT x WHERE x = :x_1", checkparams={"x_1": 10}
)
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
eq_(s1key.key, s2key.key)
eq_([b.value for b in s1key.bindparams], [5])
eq_([b.value for b in s2key.bindparams], [10])
def test_stmt_lambda_track_closure_binds_two(self):
def go(col_expr, whereclause, x, y):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt += lambda stmt: stmt.where(whereclause).where(
and_(c1 == x, c1 < y)
)
return stmt
c1 = column("x")
s1 = go(c1, c1 == 5, 8, 9)
s2 = go(c1, c1 == 10, 12, 14)
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
self.assert_compile(
s1,
"SELECT x WHERE x = :x_1 AND x = :x_2 AND x < :y_1",
checkparams={"x_1": 5, "x_2": 8, "y_1": 9},
)
self.assert_compile(
s2,
"SELECT x WHERE x = :x_1 AND x = :x_2 AND x < :y_1",
checkparams={"x_1": 10, "x_2": 12, "y_1": 14},
)
eq_([b.value for b in s1key.bindparams], [5, 8, 9])
eq_([b.value for b in s2key.bindparams], [10, 12, 14])
s1_compiled_cached = s1.compile(cache_key=s1key)
params = s1_compiled_cached.construct_params(
extracted_parameters=s2key[1]
)
eq_(params, {"x_1": 10, "x_2": 12, "y_1": 14})
def test_stmt_lambda_track_closure_binds_three(self):
def go(col_expr, whereclause, x, y):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt += lambda stmt: stmt.where(whereclause)
stmt += lambda stmt: stmt.where(and_(c1 == x, c1 < y))
return stmt
c1 = column("x")
s1 = go(c1, c1 == 5, 8, 9)
s2 = go(c1, c1 == 10, 12, 14)
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
self.assert_compile(
s1,
"SELECT x WHERE x = :x_1 AND x = :x_2 AND x < :y_1",
checkparams={"x_1": 5, "x_2": 8, "y_1": 9},
)
self.assert_compile(
s2,
"SELECT x WHERE x = :x_1 AND x = :x_2 AND x < :y_1",
checkparams={"x_1": 10, "x_2": 12, "y_1": 14},
)
eq_([b.value for b in s1key.bindparams], [5, 8, 9])
eq_([b.value for b in s2key.bindparams], [10, 12, 14])
s1_compiled_cached = s1.compile(cache_key=s1key)
params = s1_compiled_cached.construct_params(
extracted_parameters=s2key[1]
)
eq_(params, {"x_1": 10, "x_2": 12, "y_1": 14})
def test_stmt_lambda_w_atonce_whereclause_novalue(self):
def go(col_expr, whereclause):
stmt = lambdas.lambda_stmt(lambda: select(col_expr))
stmt += lambda stmt: stmt.where(whereclause)
return stmt
c1 = column("x")
s1 = go(c1, bindparam("x"))
self.assert_compile(s1, "SELECT x WHERE :x")
def test_stmt_lambda_w_additional_hashable_variants(self):
# note a Python 2 old style class would fail here because it
# isn't hashable. right now we do a hard check for __hash__ which
# will raise if the attr isn't present
class Thing(object):
def __init__(self, col_expr):
self.col_expr = col_expr
def go(thing, q):
stmt = lambdas.lambda_stmt(lambda: select(thing.col_expr))
stmt += lambda stmt: stmt.where(thing.col_expr == q)
return stmt
c1 = Thing(column("x"))
c2 = Thing(column("y"))
s1 = go(c1, 5)
s2 = go(c2, 10)
s3 = go(c1, 8)
s4 = go(c2, 12)
self.assert_compile(
s1, "SELECT x WHERE x = :q_1", checkparams={"q_1": 5}
)
self.assert_compile(
s2, "SELECT y WHERE y = :q_1", checkparams={"q_1": 10}
)
self.assert_compile(
s3, "SELECT x WHERE x = :q_1", checkparams={"q_1": 8}
)
self.assert_compile(
s4, "SELECT y WHERE y = :q_1", checkparams={"q_1": 12}
)
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
s3key = s3._generate_cache_key()
s4key = s4._generate_cache_key()
eq_(s1key[0], s3key[0])
eq_(s2key[0], s4key[0])
ne_(s1key[0], s2key[0])
def test_stmt_lambda_w_set_of_opts(self):
stmt = lambdas.lambda_stmt(lambda: select(column("x")))
opts = {column("x"), column("y")}
assert_raises_message(
exc.ArgumentError,
'Can\'t create a cache key for lambda closure variable "opts" '
"because it's a set. try using a list",
stmt.__add__,
lambda stmt: stmt.options(*opts),
)
def test_stmt_lambda_w_list_of_opts(self):
def go(opts):
stmt = lambdas.lambda_stmt(lambda: select(column("x")))
stmt += lambda stmt: stmt.options(*opts)
return stmt
s1 = go([column("a"), column("b")])
s2 = go([column("a"), column("b")])
s3 = go([column("q"), column("b")])
s1key = s1._generate_cache_key()
s2key = s2._generate_cache_key()
s3key = s3._generate_cache_key()
eq_(s1key.key, s2key.key)
ne_(s1key.key, s3key.key)
def test_stmt_lambda_hey_theres_multiple_paths(self):
def go(x, y):
stmt = lambdas.lambda_stmt(lambda: select(column("x")))
if x > 5:
stmt += lambda stmt: stmt.where(column("x") == x)
else:
stmt += lambda stmt: stmt.where(column("y") == y)
stmt += lambda stmt: stmt.order_by(column("q"))
# TODO: need more path variety here to exercise
# using a full path key
return stmt
s1 = go(2, 5)
s2 = go(8, 7)
s3 = go(4, 9)
s4 = go(10, 1)
self.assert_compile(s1, "SELECT x WHERE y = :y_1 ORDER BY q")
self.assert_compile(s2, "SELECT x WHERE x = :x_1 ORDER BY q")
self.assert_compile(s3, "SELECT x WHERE y = :y_1 ORDER BY q")
self.assert_compile(s4, "SELECT x WHERE x = :x_1 ORDER BY q")
def test_coercion_cols_clause(self):
assert_raises_message(
exc.ArgumentError,
"Textual column expression 'f' should be explicitly declared",
select,
[lambda: "foo"],
)
def test_coercion_where_clause(self):
assert_raises_message(
exc.ArgumentError,
"SQL expression for WHERE/HAVING role expected, got 5",
select(column("q")).where,
5,
)
def test_propagate_attrs_full_stmt(self):
col = column("q")
col._propagate_attrs = col._propagate_attrs.union(
{"compile_state_plugin": "x", "plugin_subject": "y"}
)
stmt = lambdas.lambda_stmt(lambda: select(col))
eq_(
stmt._propagate_attrs,
{"compile_state_plugin": "x", "plugin_subject": "y"},
)
def test_propagate_attrs_cols_clause(self):
col = column("q")
col._propagate_attrs = col._propagate_attrs.union(
{"compile_state_plugin": "x", "plugin_subject": "y"}
)
stmt = select(lambda: col)
eq_(
stmt._propagate_attrs,
{"compile_state_plugin": "x", "plugin_subject": "y"},
)
def test_propagate_attrs_from_clause(self):
col = column("q")
t = table("t", column("y"))
t._propagate_attrs = t._propagate_attrs.union(
{"compile_state_plugin": "x", "plugin_subject": "y"}
)
stmt = future_select(lambda: col).join(t)
eq_(
stmt._propagate_attrs,
{"compile_state_plugin": "x", "plugin_subject": "y"},
)
def test_select_legacy_expanding_columns(self):
q, p, r = column("q"), column("p"), column("r")
stmt = select(lambda: (q, p, r))
self.assert_compile(stmt, "SELECT q, p, r")
def test_select_future_expanding_columns(self):
q, p, r = column("q"), column("p"), column("r")
stmt = future_select(lambda: (q, p, r))
self.assert_compile(stmt, "SELECT q, p, r")
def test_select_fromclause(self):
t1 = table("t1", column("q"), column("p"))
t2 = table("t2", column("y"))
def go():
return select(t1).select_from(
lambda: join(t1, t2, lambda: t1.c.q == t2.c.y)
)
self.assert_compile(
go(), "SELECT t1.q, t1.p FROM t1 JOIN t2 ON t1.q = t2.y"
)
self.assert_compile(
go(), "SELECT t1.q, t1.p FROM t1 JOIN t2 ON t1.q = t2.y"
)
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 (:q_1_1, :q_1_2, :q_1_3)",
render_postcompile=True,
checkparams={"q_1_1": "a", "q_1_2": "b", "q_1_3": "c"},
)
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 (:q_1_1, :q_1_2, :q_1_3)",
render_postcompile=True,
checkparams={"q_1_1": "a", "q_1_2": "b", "q_1_3": "c"},
)
def test_in_parameters_three(self):
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 (:q_1_1, :q_1_2, :q_1_3)",
render_postcompile=True,
checkparams={"q_1_1": "a", "q_1_2": "b", "q_1_3": "c"},
)
def test_in_parameters_four(self):
def go(names):
return lambdas.lambda_stmt(
lambda: select(1).where(column("q").in_(names))
)
expr4 = go(["a", "b", "c"])
self.assert_compile(
expr4, "SELECT 1 WHERE q IN ([POSTCOMPILE_names_1])"
)
self.assert_compile(
expr4,
"SELECT 1 WHERE q IN (:names_1_1, :names_1_2, :names_1_3)",
render_postcompile=True,
checkparams={"names_1_1": "a", "names_1_2": "b", "names_1_3": "c"},
)
def test_in_parameters_five(self):
def go(n1, n2):
stmt = lambdas.lambda_stmt(
lambda: select(1).where(column("q").in_(n1))
)
stmt += lambda s: s.where(column("y").in_(n2))
return stmt
expr = go(["a", "b", "c"], ["d", "e", "f"])
self.assert_compile(
expr,
"SELECT 1 WHERE q IN (:n1_1_1, :n1_1_2, :n1_1_3) "
"AND y IN (:n2_1_1, :n2_1_2, :n2_1_3)",
render_postcompile=True,
checkparams={
"n1_1_1": "a",
"n1_1_2": "b",
"n1_1_3": "c",
"n2_1_1": "d",
"n2_1_2": "e",
"n2_1_3": "f",
},
)
def test_select_columns_clause(self):
t1 = table("t1", column("q"), column("p"))
g = 5
def go():
return select(lambda: t1.c.q, lambda: t1.c.p + g)
stmt = go()
self.assert_compile(
stmt,
"SELECT t1.q, t1.p + :g_1 AS anon_1 FROM t1",
checkparams={"g_1": 5},
)
eq_(stmt._generate_cache_key()._generate_param_dict(), {"g_1": 5})
g = 10
stmt = go()
self.assert_compile(
stmt,
"SELECT t1.q, t1.p + :g_1 AS anon_1 FROM t1",
checkparams={"g_1": 10},
)
eq_(stmt._generate_cache_key()._generate_param_dict(), {"g_1": 10})
@testing.metadata_fixture()
def user_address_fixture(self, metadata):
users = Table(
"users",
metadata,
Column("id", Integer, primary_key=True),
Column("name", String(50)),
)
addresses = Table(
"addresses",
metadata,
Column("id", Integer),
Column("user_id", ForeignKey("users.id")),
Column("email", String(50)),
)
return users, addresses
def test_adapt_select(self, user_address_fixture):
users, addresses = user_address_fixture
stmt = (
select(users)
.select_from(
users.join(
addresses, lambda: users.c.id == addresses.c.user_id
)
)
.where(lambda: users.c.name == "ed")
)
self.assert_compile(
stmt,
"SELECT users.id, users.name FROM users "
"JOIN addresses ON users.id = addresses.user_id "
"WHERE users.name = :name_1",
)
u1 = users.alias()
adapter = sql_util.ClauseAdapter(u1)
s2 = adapter.traverse(stmt)
self.assert_compile(
s2,
"SELECT users_1.id, users_1.name FROM users AS users_1 "
"JOIN addresses ON users_1.id = addresses.user_id "
"WHERE users_1.name = :name_1",
)
def test_no_var_dict_keys(self, user_address_fixture):
users, addresses = user_address_fixture
names = {"x": "some name"}
foo = "x"
expr = lambda: users.c.name == names[foo] # noqa
assert_raises_message(
exc.InvalidRequestError,
"Dictionary keys / list indexes inside of a cached "
"lambda must be Python literals only",
coercions.expect,
roles.WhereHavingRole,
expr,
)
def test_dict_literal_keys(self, user_address_fixture):
users, addresses = user_address_fixture
names = {"x": "some name"}
lmb = lambda: users.c.name == names["x"] # noqa
expr = coercions.expect(roles.WhereHavingRole, lmb)
self.assert_compile(
expr,
"users.name = :x_1",
params=expr._param_dict(),
checkparams={"x_1": "some name"},
)
def test_assignment_one(self, user_address_fixture):
users, addresses = user_address_fixture
x = 5
def my_lambda():
y = 10
z = y + 18
expr1 = users.c.name > x
expr2 = users.c.name < z
return and_(expr1, expr2)
expr = coercions.expect(roles.WhereHavingRole, my_lambda)
self.assert_compile(
expr,
"users.name > :x_1 AND users.name < :name_1",
params=expr._param_dict(),
checkparams={"name_1": 28, "x_1": 5},
)
expr = coercions.expect(roles.WhereHavingRole, my_lambda)
self.assert_compile(
expr,
"users.name > :x_1 AND users.name < :name_1",
params=expr._param_dict(),
checkparams={"name_1": 28, "x_1": 5},
)
def test_assignment_two(self, user_address_fixture):
users, addresses = user_address_fixture
x = 5
z = 10
def my_lambda():
y = x + z
expr1 = users.c.name > x
expr2 = users.c.name < y
return and_(expr1, expr2)
expr = coercions.expect(roles.WhereHavingRole, my_lambda)
self.assert_compile(
expr,
"users.name > :x_1 AND users.name < :x_1 + :z_1",
params=expr._param_dict(),
checkparams={"x_1": 5, "z_1": 10},
)
x = 15
z = 18
expr = coercions.expect(roles.WhereHavingRole, my_lambda)
self.assert_compile(
expr,
"users.name > :x_1 AND users.name < :x_1 + :z_1",
params=expr._param_dict(),
checkparams={"x_1": 15, "z_1": 18},
)
def test_assignment_three(self, user_address_fixture):
users, addresses = user_address_fixture
x = 5
z = 10
def my_lambda():
y = 10 + z
expr1 = users.c.name > x
expr2 = users.c.name < y
return and_(expr1, expr2)
expr = coercions.expect(roles.WhereHavingRole, my_lambda)
self.assert_compile(
expr,
"users.name > :x_1 AND users.name < :param_1 + :z_1",
params=expr._param_dict(),
checkparams={"x_1": 5, "z_1": 10, "param_1": 10},
)
x = 15
z = 18
expr = coercions.expect(roles.WhereHavingRole, my_lambda)
self.assert_compile(
expr,
"users.name > :x_1 AND users.name < :param_1 + :z_1",
params=expr._param_dict(),
checkparams={"x_1": 15, "z_1": 18, "param_1": 10},
)
def test_op_reverse(self, user_address_fixture):
user, addresses = user_address_fixture
x = "foo"
def mylambda():
return x + user.c.name
expr = coercions.expect(roles.WhereHavingRole, mylambda)
self.assert_compile(
expr, ":x_1 || users.name", checkparams={"x_1": "foo"}
)
x = "bar"
expr = coercions.expect(roles.WhereHavingRole, mylambda)
self.assert_compile(
expr, ":x_1 || users.name", checkparams={"x_1": "bar"}
)
def test_op_forwards(self, user_address_fixture):
user, addresses = user_address_fixture
x = "foo"
def mylambda():
return user.c.name + x
expr = coercions.expect(roles.WhereHavingRole, mylambda)
self.assert_compile(
expr, "users.name || :x_1", checkparams={"x_1": "foo"}
)
x = "bar"
expr = coercions.expect(roles.WhereHavingRole, mylambda)
self.assert_compile(
expr, "users.name || :x_1", checkparams={"x_1": "bar"}
)
def test_execute_constructed_uncached(self, user_address_fixture):
users, addresses = user_address_fixture
def go(name):
stmt = select(lambda: users.c.id).where(
lambda: users.c.name == name
)
with testing.db.connect().execution_options(
compiled_cache=None
) as conn:
conn.execute(stmt)
with self.sql_execution_asserter(testing.db) as asserter:
go("name1")
go("name2")
go("name1")
go("name3")
asserter.assert_(
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name1"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name2"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name1"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name3"}],
),
)
def test_execute_full_uncached(self, user_address_fixture):
users, addresses = user_address_fixture
def go(name):
stmt = lambda_stmt(
lambda: select(users.c.id).where(users.c.name == name) # noqa
)
with testing.db.connect().execution_options(
compiled_cache=None
) as conn:
conn.execute(stmt)
with self.sql_execution_asserter(testing.db) as asserter:
go("name1")
go("name2")
go("name1")
go("name3")
asserter.assert_(
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name1"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name2"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name1"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name3"}],
),
)
def test_execute_constructed_cached(self, user_address_fixture):
users, addresses = user_address_fixture
cache = {}
def go(name):
stmt = select(lambda: users.c.id).where(
lambda: users.c.name == name
)
with testing.db.connect().execution_options(
compiled_cache=cache
) as conn:
conn.execute(stmt)
with self.sql_execution_asserter(testing.db) as asserter:
go("name1")
go("name2")
go("name1")
go("name3")
asserter.assert_(
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name1"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name2"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name1"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name3"}],
),
)
def test_execute_full_cached(self, user_address_fixture):
users, addresses = user_address_fixture
cache = {}
def go(name):
stmt = lambda_stmt(
lambda: select(users.c.id).where(users.c.name == name) # noqa
)
with testing.db.connect().execution_options(
compiled_cache=cache
) as conn:
conn.execute(stmt)
with self.sql_execution_asserter(testing.db) as asserter:
go("name1")
go("name2")
go("name1")
go("name3")
asserter.assert_(
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name1"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name2"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name1"}],
),
CompiledSQL(
"SELECT users.id FROM users WHERE users.name = :name_1",
lambda ctx: [{"name_1": "name3"}],
),
)
def test_cache_key_thing(self):
t1 = table("t1", column("q"), column("p"))
def go(x):
return coercions.expect(roles.WhereHavingRole, lambda: t1.c.q == x)
expr1 = go(5)
expr2 = go(10)
is_(expr1._generate_cache_key().bindparams[0], expr1._resolved.right)
is_(expr2._generate_cache_key().bindparams[0], expr2._resolved.right)
Reference in New Issue View Git Blame Copy Permalink