Add case-insensitive unique constraints (#1738)

* feat: add case-insensitive unique constraints

Add IndexColumns(case_insensitive=True) to build a case-insensitive
unique index by wrapping each named column in LOWER(). IndexColumns now
also accepts SQLAlchemy expressions for arbitrary functional indexes,
and index auto-naming derives names from the index expressions so
functional indexes get valid names.

Closes #615

* test: enforce and document functional unique index with UPPER

Add a runtime test that a raw SQLAlchemy expression functional unique
index (UPPER(name)) enforces case-insensitive uniqueness, and switch the
standalone-expression example from LOWER to UPPER in docs and test so it
is distinct from the built-in case_insensitive=True path.

Author: collerek

docs/models/index.md

@@ -359,6 +359,36 @@ You can set this parameter by providing `ormar_config` object `constraints` argu
     To set one column index use [`unique`](../fields/common-parameters.md#index) common parameter. 
     Of course, you can set many columns as indexes with this param but each of them will be a separate index.
 
+##### Case-insensitive unique index
+
+To enforce uniqueness while ignoring case (for example so that `John` and `JOHN`
+cannot both exist), pass `case_insensitive=True` together with `unique=True`. Ormar
+wraps each named column in `LOWER()` and builds a unique functional index:
+
+```python
+import ormar
+
+class Person(ormar.Model):
+    ormar_config = base_ormar_config.copy(
+        constraints=[
+            ormar.IndexColumns(
+                "first_name", "last_name", unique=True, case_insensitive=True
+            )
+        ],
+    )
+
+    id: int = ormar.Integer(primary_key=True)
+    first_name: str = ormar.String(max_length=100)
+    last_name: str = ormar.String(max_length=100)
+```
+
+!!!note
+    Case-insensitive uniqueness is a unique *functional index*, not a `UniqueColumns`
+    constraint, because a SQL unique constraint cannot hold expressions like `LOWER()`.
+    For other functional indexes you can pass SQLAlchemy expressions directly, e.g.
+    `ormar.IndexColumns(sqlalchemy.func.upper(sqlalchemy.column("name")), unique=True)`.
+    On MySQL this requires version 8.0.13 or newer.
+
 #### CheckColumns
 
 You can set this parameter by providing `ormar_config` object `constraints` argument.

docs/releases.md

@@ -1,5 +1,15 @@
 # Release notes
 
+## Unreleased
+
+### ✨ Features
+
+* Add `IndexColumns(..., case_insensitive=True)` to build a case-insensitive
+  unique index (each named column wrapped in `LOWER()`), so values that differ
+  only in case cannot both be stored. `IndexColumns` now also accepts SQLAlchemy
+  expressions for arbitrary functional indexes. On MySQL this needs 8.0.13+.
+  [#615](https://github.com/collerek/ormar/issues/615)
+
 ## 0.26.0
 
 ### ✨ Features

ormar/fields/constraints.py

@@ -1,6 +1,6 @@
 from typing import Any, Optional
 
-from sqlalchemy import CheckConstraint, Index, UniqueConstraint
+from sqlalchemy import CheckConstraint, Index, UniqueConstraint, column, func
 
 
 class UniqueColumns(UniqueConstraint):
@@ -11,16 +11,40 @@ class UniqueColumns(UniqueConstraint):
 
 
 class IndexColumns(Index):
-    def __init__(self, *args: Any, name: Optional[str] = None, **kw: Any) -> None:
-        if not name:
-            name = "TEMPORARY_NAME"
-        super().__init__(name, *args, **kw)
-
     """
     Subclass of sqlalchemy.Index.
     Used to avoid importing anything from sqlalchemy by user.
+
+    Passing ``unique=True`` together with ``case_insensitive=True`` builds a
+    case-insensitive unique index by wrapping each named column in ``LOWER()``,
+    which is how case-insensitive uniqueness is expressed (a regular
+    UniqueConstraint cannot hold SQL expressions).
     """
 
+    def __init__(
+        self,
+        *args: Any,
+        name: Optional[str] = None,
+        case_insensitive: bool = False,
+        **kw: Any,
+    ) -> None:
+        """
+        :param args: column names, or SQL expressions for a functional index
+        :type args: Any
+        :param name: optional explicit index name
+        :type name: Optional[str]
+        :param case_insensitive: wrap each named column in ``LOWER()`` so the
+            index (and its uniqueness, when ``unique=True``) ignores case
+        :type case_insensitive: bool
+        :param kw: additional keyword arguments passed to sqlalchemy.Index
+        :type kw: Any
+        """
+        if case_insensitive:
+            args = tuple(func.lower(column(arg)) for arg in args)
+        if not name:
+            name = "TEMPORARY_NAME"
+        super().__init__(name, *args, **kw)
+
 
 class CheckColumns(CheckConstraint):
     """

ormar/models/helpers/sqlalchemy.py

@@ -1,5 +1,6 @@
 import logging
-from typing import TYPE_CHECKING, ForwardRef, Optional, Union
+import re
+from typing import TYPE_CHECKING, Any, ForwardRef, Optional, Union
 
 import sqlalchemy
 
@@ -363,6 +364,20 @@ def populate_config_sqlalchemy_table_if_required(config: "OrmarConfig") -> None:
         config.table = table
 
 
+def index_name_fragment(expression: Any) -> str:
+    """
+    Returns an identifier-safe name fragment for an index column or SQL
+    expression, so auto-generated index names work for functional indexes
+    (e.g. LOWER(name)) as well as plain columns.
+
+    :param expression: an index column name or SQL expression element
+    :type expression: Any
+    :return: sanitized fragment containing only alphanumerics and underscores
+    :rtype: str
+    """
+    return re.sub(r"[^0-9a-zA-Z]+", "_", str(expression)).strip("_")
+
+
 def set_constraint_names(config: "OrmarConfig") -> None:
     """
     Populates the names on IndexColumns and UniqueColumns and CheckColumns constraints.
@@ -382,7 +397,7 @@ def set_constraint_names(config: "OrmarConfig") -> None:
         ):
             constraint.name = (
                 f"ix_{config.tablename}_"
-                f"{'_'.join([col for col in constraint._pending_colargs])}"
+                f"{'_'.join(index_name_fragment(e) for e in constraint.expressions)}"
             )
         elif isinstance(constraint, sqlalchemy.CheckConstraint) and not constraint.name:
             sql_condition: str = str(constraint.sqltext).replace(" ", "_")

tests/test_meta_constraints/test_case_insensitive_constraints.py

@@ -0,0 +1,77 @@
+import pytest
+import sqlalchemy
+
+import ormar.fields.constraints
+from tests.lifespan import init_tests
+from tests.settings import create_config
+
+base_ormar_config = create_config()
+
+
+class Person(ormar.Model):
+    ormar_config = base_ormar_config.copy(
+        tablename="persons",
+        constraints=[
+            ormar.fields.constraints.IndexColumns(
+                "first_name", "last_name", unique=True, case_insensitive=True
+            )
+        ],
+    )
+
+    id: int = ormar.Integer(primary_key=True)
+    first_name: str = ormar.String(max_length=100)
+    last_name: str = ormar.String(max_length=100)
+
+
+class Tag(ormar.Model):
+    ormar_config = base_ormar_config.copy(
+        tablename="tags",
+        constraints=[
+            ormar.fields.constraints.IndexColumns(
+                sqlalchemy.func.upper(sqlalchemy.column("name")), unique=True
+            )
+        ],
+    )
+
+    id: int = ormar.Integer(primary_key=True)
+    name: str = ormar.String(max_length=100)
+
+
+create_test_database = init_tests(base_ormar_config)
+
+
+def test_case_insensitive_unique_index_structure():
+    indexes = list(Person.ormar_config.table.indexes)
+    assert len(indexes) == 1
+    index = indexes[0]
+    assert index.unique is True
+    assert index.name == "ix_persons_lower_first_name_lower_last_name"
+
+
+def test_functional_index_expression_gets_safe_autoname():
+    indexes = list(Tag.ormar_config.table.indexes)
+    assert len(indexes) == 1
+    index = indexes[0]
+    assert index.unique is True
+    assert index.name == "ix_tags_upper_name"
+
+
+@pytest.mark.asyncio
+async def test_case_insensitive_uniqueness_is_enforced():
+    async with base_ormar_config.database:
+        async with base_ormar_config.database.transaction(force_rollback=True):
+            await Person.objects.create(first_name="John", last_name="Doe")
+            await Person.objects.create(first_name="Jane", last_name="Doe")
+
+            with pytest.raises(sqlalchemy.exc.IntegrityError):
+                await Person.objects.create(first_name="JOHN", last_name="DOE")
+
+
+@pytest.mark.asyncio
+async def test_functional_expression_uniqueness_is_enforced():
+    async with base_ormar_config.database:
+        async with base_ormar_config.database.transaction(force_rollback=True):
+            await Tag.objects.create(name="Python")
+
+            with pytest.raises(sqlalchemy.exc.IntegrityError):
+                await Tag.objects.create(name="PYTHON")