Skip to content
Merged
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
30 changes: 30 additions & 0 deletions docs/models/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
10 changes: 10 additions & 0 deletions docs/releases.md
Original file line number Diff line number Diff line change
@@ -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
Expand Down
36 changes: 30 additions & 6 deletions ormar/fields/constraints.py
Original file line number Diff line number Diff line change
@@ -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):
Expand All @@ -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):
"""
Expand Down
19 changes: 17 additions & 2 deletions ormar/models/helpers/sqlalchemy.py
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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.
Expand All @@ -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(" ", "_")
Expand Down
77 changes: 77 additions & 0 deletions tests/test_meta_constraints/test_case_insensitive_constraints.py
Original file line number Diff line number Diff line change
@@ -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")