Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: in-operator literal binds not handled properly #285

2 changes: 1 addition & 1 deletion sqlalchemy_bigquery/base.py
Expand Up @@ -420,7 +420,7 @@ def visit_bindparam(
)

type_ = bindparam.type
if isinstance(type_, NullType):
if literal_binds or isinstance(type_, NullType):
return param

if (
Expand Down
2 changes: 1 addition & 1 deletion testing/constraints-3.6.txt
Expand Up @@ -6,5 +6,5 @@
# e.g., if setup.py has "foo >= 1.14.0, < 2.0.0dev",
sqlalchemy==1.2.0
google-auth==1.25.0
google-cloud-bigquery==2.19.0
google-cloud-bigquery==2.24.1
google-api-core==1.30.0
16 changes: 16 additions & 0 deletions tests/unit/test_compliance.py
Expand Up @@ -200,3 +200,19 @@ def test_group_by_composed(faux_conn):
select([sqlalchemy.func.count(table.c.id), expr]).group_by(expr).order_by(expr)
)
assert_result(faux_conn, stmt, [(1, 3), (1, 5), (1, 7)])


def test_cast_type_decorator(faux_conn, last_query):
# [artial dup of:
# sqlalchemy.testing.suite.test_types.CastTypeDecoratorTest.test_special_type
# That test failes without code that's otherwise not covered by the unit tests.

class StringAsInt(sqlalchemy.TypeDecorator):
impl = sqlalchemy.String(50)

def bind_expression(self, col):
return sqlalchemy.cast(col, String(50))

t = setup_table(faux_conn, "t", Column("x", StringAsInt()))
faux_conn.execute(t.insert(), [{"x": x} for x in [1, 2, 3]])
last_query("INSERT INTO `t` (`x`) VALUES (CAST(%(x:STRING)s AS STRING))", {"x": 3})
18 changes: 18 additions & 0 deletions tests/unit/test_select.py
Expand Up @@ -356,3 +356,21 @@ def test_select_notin_param_empty(faux_conn):
else "SELECT (%(param_1:INT64)s NOT IN UNNEST([ ])) AS `anon_1`",
{"param_1": 1},
)


def test_literal_binds_kwarg_with_an_IN_operator_252(faux_conn):
table = setup_table(
faux_conn,
"test",
sqlalchemy.Column("val", sqlalchemy.Integer),
initial_data=[dict(val=i) for i in range(3)],
)
q = sqlalchemy.select([table.c.val]).where(table.c.val.in_([2]))

def nstr(q):
return " ".join(str(q).strip().split())

assert (
nstr(q.compile(faux_conn.engine, compile_kwargs={"literal_binds": True}))
== "SELECT `test`.`val` FROM `test` WHERE `test`.`val` IN (2)"
)