Add __like and __ilike filters (#421)

CHANGELOG.rst

@@ -10,6 +10,13 @@ Changelog
 0.25
 ====
 
+0.25.2 (Unreleased)
+------
+
+Added
+^^^^^
+- Add `__like` and `__ilike` filters. (#421)
+
 0.25.1
 ------------------
 Changed

docs/query.rst

@@ -215,6 +215,8 @@ The following lookup types are available:
 - ``not_isnull`` - field is not null
 - ``contains`` - field contains specified substring
 - ``icontains`` - case insensitive ``contains``
+- ``like`` - field matches the specified pattern (may contain the SQL wildcards ``%`` and ``_``)
+- ``ilike`` - case insensitive ``like``
 - ``startswith`` - if field starts with value
 - ``istartswith`` - case insensitive ``startswith``
 - ``endswith`` - if field ends with value

examples/transactions.py

@@ -2,6 +2,8 @@
 This example demonstrates how you can use transactions with tortoise
 """
 
+import contextlib
+
 from tortoise import Tortoise, fields, run_async
 from tortoise.exceptions import OperationalError
 from tortoise.models import Model
@@ -43,10 +45,8 @@ async def bound_to_fall():
         print(saved_event.name)
         raise OperationalError()
 
-    try:
+    with contextlib.suppress(OperationalError):
         await bound_to_fall()
-    except OperationalError:
-        pass
     saved_event = await Event.filter(name="Updated name").first()
     print(saved_event)
 

pyproject.toml

@@ -138,36 +138,11 @@ show_error_context = true
 [[tool.mypy.overrides]]
 module = "tests.*"
 check_untyped_defs = false
-disallow_untyped_defs = false
-disallow_incomplete_defs = false
-warn_unreachable = false
 
 [[tool.mypy.overrides]]
 module = "examples.*"
 check_untyped_defs = false
 disallow_untyped_calls = false
-disallow_untyped_defs = false
-disallow_incomplete_defs = false
-
-[[tool.mypy.overrides]]
-module = "examples.fastapi.*"
-check_untyped_defs = true
-disallow_untyped_calls = true
-disallow_untyped_defs = false
-disallow_incomplete_defs = false
-
-[[tool.mypy.overrides]]
-module = "tortoise.contrib.test.*"
-disallow_untyped_defs = false
-disallow_incomplete_defs = false
-
-[[tool.mypy.overrides]]
-module = "tortoise.contrib.sanic.*"
-disallow_untyped_defs = false
-
-[[tool.mypy.overrides]]
-module = ["conftest", "tortoise.backends.base.client"]
-disallow_untyped_defs = false
 
 [tool.flake8]
 ignore = "E203,E501,W503,DAR101,DAR201,DAR402"
@@ -190,16 +165,27 @@ filterwarnings = [
 
 [tool.coverage.report]
 show_missing = true
+exclude_also = [
+    "if TYPE_CHECKING:"
+]
 
 [tool.ruff]
 line-length = 100
 [tool.ruff.lint]
 ignore = ["E501"]
 extend-select = [
+    "I",      # https://docs.astral.sh/ruff/rules/#isort-i
+    "SIM",    # https://docs.astral.sh/ruff/rules/#flake8-simplify-sim
     "FA",     # https://docs.astral.sh/ruff/rules/#flake8-future-annotations-fa
     "UP",     # https://docs.astral.sh/ruff/rules/#pyupgrade-up
     "RUF100", # https://docs.astral.sh/ruff/rules/#ruff-specific-rules-ruf
 ]
 
+[tool.ruff.lint.isort]
+extra-standard-library = ["tomllib"]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/*.py" = ["SIM117"]
+
 [tool.bandit]
 exclude_dirs = ["tests", 'examples/*/_tests.py', "conftest.py"]

tests/backends/test_mysql.py

@@ -2,6 +2,7 @@
 Test some mysql-specific features
 """
 
+import contextlib
 import ssl
 
 from tortoise import Tortoise
@@ -46,7 +47,5 @@ async def test_ssl_custom(self):
         ctx.verify_mode = ssl.CERT_NONE
 
         self.db_config["connections"]["models"]["credentials"]["ssl"] = ctx
-        try:
+        with contextlib.suppress(ConnectionError):
             await Tortoise.init(self.db_config, _create_db=True)
-        except ConnectionError:
-            pass

tests/backends/test_postgres.py

@@ -2,6 +2,7 @@
 Test some PostgreSQL-specific features
 """
 
+import contextlib
 import ssl
 
 from tests.testmodels import Tournament
@@ -82,10 +83,8 @@ async def test_ssl_custom(self):
         ctx.verify_mode = ssl.CERT_NONE
 
         self.db_config["connections"]["models"]["credentials"]["ssl"] = ctx
-        try:
+        with contextlib.suppress(ConnectionError):
             await Tortoise.init(self.db_config, _create_db=True)
-        except ConnectionError:
-            pass
 
     async def test_application_name(self):
         self.db_config["connections"]["models"]["credentials"]["application_name"] = (

tests/test_filters.py

@@ -8,6 +8,8 @@
     CharPkModel,
     DecimalFields,
 )
+from tortoise import connections
+from tortoise.backends.mysql.executor import escape_backslash_except_wildcards
 from tortoise.contrib import test
 from tortoise.exceptions import FieldError
 from tortoise.fields.base import StrEnum
@@ -21,6 +23,23 @@ class MyStrEnum(StrEnum):
     moo = "moo"
 
 
+def test_escape_backslash_except_wildcards():
+    values = ["\n", "\\_", "\\%", "\\_\\%", "\\__"]
+    assert [escape_backslash_except_wildcards(v) for v in values] == values
+    value = "\\n"
+    escaped = "\\" * 2 + "n"
+    assert escape_backslash_except_wildcards(value) == escaped
+    value = "\\\\n"
+    escaped = "\\" * 4 + "n"
+    assert escape_backslash_except_wildcards(value) == escaped
+    value = "_\\"
+    escaped = "_" + "\\" * 2
+    assert escape_backslash_except_wildcards(value) == escaped
+    value = "\\\\_"
+    escaped = "\\" * 2 + "\\_"
+    assert escape_backslash_except_wildcards(value) == escaped
+
+
 class TestCharFieldFilters(test.TestCase):
     async def asyncSetUp(self):
         await super().asyncSetUp()
@@ -188,6 +207,182 @@ async def test_sorting(self):
         )
 
 
+class TestCharFieldLikeFilters(test.TestCase):
+    model_cls = CharFields
+
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        values1 = ["Like", "LIKE", "like", "lIke", "like1", "alike", "l*ke"]
+        values2 = ["L_ke", "l_ke", "l__e", "l%ke", "l%%e", "l\\ke", "L\\ke", "l\\\\e"]
+        values3 = ["L_%ke", "l_\\ke", "l\\_e", "l\\%e", "l%_ke", "l\\%e", "l\\_%_\\e"]
+        await self.model_cls.bulk_create(
+            [self.model_cls(char=i) for i in (*values1, *values2, *values3)]
+        )
+        self.db = connections.get("models")
+        dialect = self.db.schema_generator.DIALECT
+        if dialect == "sqlite":
+            await self.db.execute_script("PRAGMA case_sensitive_like=ON;")
+
+    async def test_like_percent(self):
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="like").values_list("char", flat=True)),
+            {"like"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="like%").values_list("char", flat=True)),
+            {"like", "like1"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="%like").values_list("char", flat=True)),
+            {"like", "alike"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="%like%").values_list("char", flat=True)),
+            {"like", "alike", "like1"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l%ke").values_list("char", flat=True)),
+            {"like", "lIke", "l*ke", "l_ke", "l%ke", "l\\ke", "l_\\ke", "l%_ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l%%ke").values_list("char", flat=True)),
+            {"like", "lIke", "l*ke", "l_ke", "l%ke", "l\\ke", "l_\\ke", "l%_ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l%ke%").values_list("char", flat=True)),
+            {"like", "lIke", "l*ke", "l_ke", "l%ke", "l\\ke", "l_\\ke", "l%_ke", "like1"},
+        )
+
+    async def test_like_underscore(self):
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="like_").values_list("char", flat=True)),
+            {"like1"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="_like").values_list("char", flat=True)),
+            {"alike"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="_like_").values_list("char", flat=True)),
+            set(),
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l_ke").values_list("char", flat=True)),
+            {
+                "like",
+                "lIke",
+                "l*ke",
+                "l_ke",
+                "l%ke",
+                "l\\ke",
+            },
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l__ke").values_list("char", flat=True)),
+            {"l_\\ke", "l%_ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l_ke_").values_list("char", flat=True)),
+            {"like1"},
+        )
+
+    async def test_like_backslash(self):
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l\\_\\ke").values_list("char", flat=True)),
+            {"l_\\ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l\\ke").values_list("char", flat=True)),
+            {"l\\ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l\\\\e").values_list("char", flat=True)),
+            {"l\\\\e"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l\\\\%e").values_list("char", flat=True)),
+            {"l\\%e"},
+        )
+        self.assertSetEqual(
+            set(
+                await CharFields.filter(char__like="l\\\\_\\%\\_\\e").values_list("char", flat=True)
+            ),
+            {"l\\_%_\\e"},
+        )
+
+    async def test_like_mix(self):
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l*ke").values_list("char", flat=True)),
+            {"l*ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="%like_").values_list("char", flat=True)),
+            {"like1"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="_like%").values_list("char", flat=True)),
+            {"alike"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="like_%").values_list("char", flat=True)),
+            {"like1"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l\\%").values_list("char", flat=True)),
+            set(),
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l_\\k%").values_list("char", flat=True)),
+            {"l_\\ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l%\\k_").values_list("char", flat=True)),
+            {"l\\ke", "l_\\ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l\\\\_%").values_list("char", flat=True)),
+            {"l\\_e", "l\\_%_\\e"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l\\\\__").values_list("char", flat=True)),
+            {"l\\_e"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="l\\\\_\\%%").values_list("char", flat=True)),
+            {"l\\_%_\\e"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="L\\_%").values_list("char", flat=True)),
+            {"L_%ke", "L_ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__like="L\\_\\%%").values_list("char", flat=True)),
+            {"L_%ke"},
+        )
+
+    async def test_ilike(self):
+        self.assertSetEqual(
+            set(await CharFields.filter(char__ilike="like").values_list("char", flat=True)),
+            {"like", "Like", "LIKE", "lIke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__ilike="like%").values_list("char", flat=True)),
+            {"like", "like1", "Like", "LIKE", "lIke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__ilike="l_ke").values_list("char", flat=True)),
+            {"LIKE", "Like", "like", "lIke", "l*ke", "l_ke", "l%ke", "l\\ke", "L\\ke", "L_ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__ilike="l\\ke").values_list("char", flat=True)),
+            {"l\\ke", "L\\ke"},
+        )
+        self.assertSetEqual(
+            set(await CharFields.filter(char__ilike="L\\_%").values_list("char", flat=True)),
+            {"L_%ke", "L_ke", "l_ke", "l_\\ke", "l__e"},
+        )
+
+
 class TestBooleanFieldFilters(test.TestCase):
     async def asyncSetUp(self):
         await super().asyncSetUp()

tests/test_only.py

@@ -1,7 +1,7 @@
 from tests.testmodels import DoubleFK, Event, SourceFields, StraightFields, Tournament
 from tortoise.contrib import test
-from tortoise.functions import Count
 from tortoise.exceptions import FieldError, IncompleteInstanceError
+from tortoise.functions import Count
 
 
 class TestOnlyStraight(test.TestCase):

tortoise/backends/asyncpg/client.py

@@ -130,10 +130,7 @@ async def execute_many(self, query: str, values: list) -> None:
     async def execute_query(self, query: str, values: list | None = None) -> tuple[int, list[dict]]:
         async with self.acquire_connection() as connection:
             self.log.debug("%s: %s", query, values)
-            if values:
-                params = [query, *values]
-            else:
-                params = [query]
+            params = [query, *values] if values else [query]
             if query.startswith("UPDATE") or query.startswith("DELETE"):
                 res = await connection.execute(*params)
                 try: