Skip to content
Permalink
Browse files
fix: in-operator literal binds not handled properly (#285)
  • Loading branch information
jimfulton committed Aug 23, 2021
1 parent 2cf05a0 commit e06bf74310fa27d5bc775e13beed4ab3a520e1aa
@@ -420,7 +420,7 @@ def visit_bindparam(
)

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

if (
@@ -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
@@ -301,7 +301,7 @@ def test_record_content_from_raw_queries(engine, bigquery_dataset):


def test_content_from_reflect(engine, table_one_row):
rows = table_one_row.select().execute().fetchall()
rows = table_one_row.select(use_labels=True).execute().fetchall()
assert list(rows[0]) == ONE_ROW_CONTENTS_EXPANDED


@@ -505,21 +505,21 @@ def test_querying_wildcard_tables(engine):
def test_dml(engine, session, table_dml):
# test insert
engine.execute(table_dml.insert(ONE_ROW_CONTENTS_DML))
result = table_dml.select().execute().fetchall()
result = table_dml.select(use_labels=True).execute().fetchall()
assert len(result) == 1

# test update
session.query(table_dml).filter(table_dml.c.string == "test").update(
{"string": "updated_row"}, synchronize_session=False
)
updated_result = table_dml.select().execute().fetchone()
updated_result = table_dml.select(use_labels=True).execute().fetchone()
assert updated_result[table_dml.c.string] == "updated_row"

# test delete
session.query(table_dml).filter(table_dml.c.string == "updated_row").delete(
synchronize_session=False
)
result = table_dml.select().execute().fetchall()
result = table_dml.select(use_labels=True).execute().fetchall()
assert len(result) == 0


@@ -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})
@@ -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)"
)

0 comments on commit e06bf74

Please sign in to comment.