Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions DESCRIPTION.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,9 @@ Source code is also available at:

# Unreleased Notes

- vX.X.X
- Add support for ILIKE in queries

# Release Notes

- v1.8.0(December 5, 2025)
Expand Down
20 changes: 19 additions & 1 deletion src/snowflake/sqlalchemy/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
from sqlalchemy.orm import context
from sqlalchemy.orm.context import _MapperEntity
from sqlalchemy.schema import Sequence, Table
from sqlalchemy.sql import compiler, expression, functions
from sqlalchemy.sql import compiler, expression, functions, sqltypes
from sqlalchemy.sql.base import CompileState
from sqlalchemy.sql.elements import BindParameter, quoted_name
from sqlalchemy.sql.expression import Executable
Expand Down Expand Up @@ -779,6 +779,24 @@ def visit_regexp_replace_op_binary(self, binary, operator, **kw):
def visit_not_regexp_match_op_binary(self, binary, operator, **kw):
return f"NOT {self.visit_regexp_match_op_binary(binary, operator, **kw)}"

def visit_ilike_op_binary(self, binary, operator, **kw):
return self._render_ilike(binary, negate=False, **kw)

def visit_not_ilike_op_binary(self, binary, operator, **kw):
return self._render_ilike(binary, negate=True, **kw)

def _render_ilike(self, binary, negate=False, **kw):
left = binary.left._compiler_dispatch(self, **kw)
right = binary.right._compiler_dispatch(self, **kw)
escape = binary.modifiers.get("escape")
escape_clause = (
" ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
if escape is not None
else ""
)
operator = "NOT ILIKE" if negate else "ILIKE"
return f"{left} {operator} {right}{escape_clause}"

def visit_join(self, join, asfrom=False, from_linter=None, **kwargs):
if from_linter:
from_linter.edges.update(
Expand Down
15 changes: 15 additions & 0 deletions tests/test_compiler.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,21 @@ def test_multi_table_update(self):
"WHERE table1.id = test.table2.name",
)

def test_ilike_compilation(self):
statement = select(table1.c.name).where(table1.c.name.ilike("%Ann%"))
self.assert_compile(
statement,
"SELECT table1.name FROM table1 WHERE table1.name ILIKE %(name_1)s",
)

statement = select(table1.c.name).where(
~table1.c.name.ilike("foo\\_%", escape="\\")
)
self.assert_compile(
statement,
"SELECT table1.name FROM table1 WHERE table1.name NOT ILIKE %(name_1)s ESCAPE '\\\\'",
)

def test_drop_table_comment(self):
self.assert_compile(DropTableComment(table1), "COMMENT ON TABLE table1 IS ''")
self.assert_compile(
Expand Down
71 changes: 57 additions & 14 deletions tests/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -420,6 +420,49 @@ def test_insert_tables(engine_testaccount):
users.drop(engine_testaccount)


def test_ilike_support(engine_testaccount):
metadata = MetaData()
table = Table(
f"ilike_test_{random_string(5)}",
metadata,
Column("id", Integer, primary_key=True),
Column("value", String),
)
metadata.create_all(engine_testaccount)

cases = [
# (query_pattern, escape, negate, expected_query_result)
("casesens%", None, False, ["CaseSensitive", "casesensitive"]),
("casesens%", None, True, ["pattern_test", "pattern_%", "patternstest"]),
("pattern_%", None, False, ["pattern_test", "pattern_%", "patternstest"]),
("pattern\\_%", "\\", False, ["pattern_test", "pattern_%"]),
("pattern\\_\\%", "\\", False, ["pattern_%"]),
]

try:
with engine_testaccount.begin() as conn:
conn.execute(
table.insert(),
[
{"id": 1, "value": "CaseSensitive"},
{"id": 2, "value": "casesensitive"},
{"id": 3, "value": "pattern_test"},
{"id": 4, "value": "pattern_%"},
{"id": 5, "value": "patternstest"},
],
)

for pattern, escape, negate, expected in cases:
clause = table.c.value.ilike(pattern, escape=escape)
if negate:
clause = ~clause

rows = conn.execute(select(table.c.value).where(clause)).scalars().all()
assert sorted(rows) == sorted(expected)
finally:
metadata.drop_all(engine_testaccount)


def test_table_does_not_exist(engine_testaccount):
"""
Tests Correct Exception Thrown When Table Does Not Exist
Expand Down Expand Up @@ -1535,20 +1578,20 @@ def test_for_exception_in_query_all_columns(engine_testaccount, db_parameters):

exception_instance = DBAPIError.instance(
"""
SELECT /* sqlalchemy:_get_schema_columns */
ic.table_name,
ic.column_name,
ic.data_type,
ic.character_maximum_length,
ic.numeric_precision,
ic.numeric_scale,
ic.is_nullable,
ic.column_default,
ic.is_identity,
ic.comment
FROM information_schema.columns ic
WHERE ic.table_schema='schema_name'
ORDER BY ic.ordinal_position""",
SELECT /* sqlalchemy:_get_schema_columns */
ic.table_name,
ic.column_name,
ic.data_type,
ic.character_maximum_length,
ic.numeric_precision,
ic.numeric_scale,
ic.is_nullable,
ic.column_default,
ic.is_identity,
ic.comment
FROM information_schema.columns ic
WHERE ic.table_schema = 'schema_name'
ORDER BY ic.ordinal_position""",
{"table_schema": "TESTSCHEMA"},
ProgrammingError(
"Information schema query returned too much data. Please repeat query with more "
Expand Down
Loading