feat: Add support validating indexes with where statement

docs/db_validation.md

@@ -24,4 +24,17 @@ If the uniqueness constraint is violated, an `fastapi.HTTPException` will be rai
 
 If the model defines unique constraints using `sqlalchemy.UniqueConstraint`, then when using the `create` or `update` methods, an SQL query will be executed to verify that no other objects with the same combination of field values in the unique constraint exist.
 
-If the unique constraint is violated, an `fastapi.HTTPException` will be raised.
+If the unique constraint is violated, an `fastapi.HTTPException` will be raised.
+
+## Unique Indexes Validation
+
+If the model defines a unique `sqlalchemy.Index`, then when using the `create` or `update` methods, an SQL query will be executed to verify that no other object violates that index.
+
+For **partial** unique indexes (`postgresql_where=...`), validation runs only when the created/updated row itself matches the `WHERE` predicate. Rows outside the partial index are not checked against it.
+
+Supported `postgresql_where` forms for deciding whether the row is covered:
+
+- Column expressions such as `Model.kind == "main"` (including `and_(...)` of equalities)
+- Simple `text()` equalities such as `text("kind = 'main'")`
+
+Unsupported / complex predicates are treated as matching (fail-safe): validation still runs.

docs/ru/db_validation.md

@@ -32,4 +32,20 @@ Ecли `Post.slug` -- поле, определённое с `unique=True`, и з
 при использовании методов `create` или `update` будет выполнен SQL запрос для проверки
 существования других объектов с таким же набором значений полей, входящих в уникальное огранчение.
 
-Если уникальное ограничение будет нарушено, вызывается `fastapi.HTTPException`.
+Если уникальное ограничение будет нарушено, вызывается `fastapi.HTTPException`.
+
+## Валидация уникальных индексов
+
+Если в модели определён уникальный `sqlalchemy.Index`, то при использовании методов
+`create` или `update` будет выполнен SQL запрос для проверки, что индекс не нарушается.
+
+Для **частичных** уникальных индексов (`postgresql_where=...`) валидация выполняется
+только если создаваемая/обновляемая запись сама попадает под условие `WHERE`.
+Записи вне partial index по этому индексу не проверяются.
+
+Поддерживаемые формы `postgresql_where` для определения покрытия записи:
+
+- выражения по колонкам, например `Model.kind == "main"` (включая `and_(...)` из равенств)
+- простые равенства через `text()`, например `text("kind = 'main'")`
+
+Неподдерживаемые / сложные предикаты считаются подходящими (fail-safe): валидация всё равно выполняется.

fastapi_sqlalchemy_toolkit/model_manager.py

@@ -1,5 +1,6 @@
 # ruff: noqa: UP006
 from collections.abc import Callable, Iterable
+from enum import Enum
 from typing import Any, Generic, List, TypeVar, overload  # noqa: UP035
 
 from fastapi import HTTPException, status
@@ -25,9 +26,16 @@
 from sqlalchemy.orm import DeclarativeBase, contains_eager, load_only
 from sqlalchemy.orm.attributes import InstrumentedAttribute
 from sqlalchemy.orm.relationships import Relationship
-from sqlalchemy.sql import Select
-from sqlalchemy.sql.elements import BindParameter, Null, UnaryExpression
-from sqlalchemy.sql.expression import BinaryExpression, ColumnElement
+from sqlalchemy.sql import Select, operators
+from sqlalchemy.sql.elements import (
+    BinaryExpression,
+    BindParameter,
+    BooleanClauseList,
+    Null,
+    TextClause,
+    UnaryExpression,
+)
+from sqlalchemy.sql.expression import ColumnElement
 from sqlalchemy.sql.functions import Function
 from sqlalchemy.sql.schema import ScalarElementColumnDefault
 from sqlalchemy.sql.selectable import Exists
@@ -1483,18 +1491,87 @@ async def validate_unique_fields(
                         ),
                     )
 
+    @staticmethod
+    def _partial_index_values_equal(actual: Any, expected: Any) -> bool:
+        """Сравнить обычные и Enum-значения из объекта и условия индекса."""
+        actual_normalized = actual.name if isinstance(actual, Enum) else actual
+        expected_normalized = expected.name if isinstance(expected, Enum) else expected
+        if actual_normalized == expected_normalized:
+            return True
+        if isinstance(actual, Enum) and actual.value == expected_normalized:
+            return True
+        if isinstance(expected, Enum) and expected.value == actual_normalized:
+            return True
+        return str(actual_normalized) == str(expected_normalized)
+
+    @classmethod
+    def _matches_partial_index_where(cls, condition: Any, in_obj: ModelDict) -> bool:
+        """
+        Проверить, покрывается ли объект условием partial index.
+
+        Неподдерживаемое условие считается совпавшим, чтобы не пропустить
+        потенциальное нарушение уникальности.
+        """
+        if isinstance(condition, BooleanClauseList):
+            return all(
+                cls._matches_partial_index_where(clause, in_obj) for clause in condition
+            )
+
+        column_name: str | None = None
+        expected: Any = None
+        if isinstance(condition, TextClause):
+            raw_column, separator, raw_expected = condition.text.partition("=")
+            column_name = raw_column.strip()
+            raw_expected = raw_expected.strip()
+            valid_column = column_name.isascii() and column_name.isidentifier()
+
+            if not separator or not valid_column or not raw_expected:
+                column_name = None
+            elif raw_expected[0] in {'"', "'"}:
+                quote = raw_expected[0]
+                if raw_expected[-1] == quote and quote not in raw_expected[1:-1]:
+                    expected = raw_expected[1:-1]
+                else:
+                    column_name = None
+            elif all(char.isalnum() or char in "._" for char in raw_expected):
+                expected = raw_expected
+            else:
+                column_name = None
+        elif (
+            isinstance(condition, BinaryExpression)
+            and condition.operator is operators.eq
+        ):
+            column_name = getattr(condition.left, "name", None) or getattr(
+                condition.left, "key", None
+            )
+            expected = getattr(
+                condition.right,
+                "value",
+                getattr(condition.right, "effective_value", condition.right),
+            )
+
+        if not isinstance(column_name, str) or column_name not in in_obj:
+            return True
+        return cls._partial_index_values_equal(in_obj[column_name], expected)
+
     async def validate_unique_indexes(
         self, session: AsyncSession, in_obj: ModelDict
     ) -> None:
         """
-        Валидирует соблюдение уникальности индексов
+        Валидирует соблюдение уникальности индексов.
+
+        Для partial unique index (`postgresql_where`) проверка выполняется только
+        если создаваемая/обновляемая запись сама попадает под WHERE.
         """
         for index in self.unique_indexes:
             condition = (
                 index.dialect_options["postgresql"].get("where")
                 if index.dialect_options.get("postgresql")
                 else None
             )
+            if not self._matches_partial_index_where(condition, in_obj):
+                continue
+
             filters = []
 
             for column in index.columns:

requirements/test.txt

@@ -1,2 +1,2 @@
-pytest>=8.0.0
+pytest>=9.0.0
 asyncpg>=0.29.0

tests/conftest.py

@@ -1,4 +1,4 @@
-from typing import AsyncGenerator
+from collections.abc import AsyncGenerator
 
 import pytest
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession, AsyncTransaction
@@ -7,25 +7,35 @@
 from tests.models import Base, CustomPKBase
 
 
-@pytest.fixture(autouse=True)
-async def create_metadata():
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
-        await conn.run_sync(CustomPKBase.metadata.create_all)
-
-
 @pytest.fixture(scope="session")
-def anyio_backend():
+def anyio_backend() -> str:
     return "asyncio"
 
 
 @pytest.fixture(scope="session")
-async def connection(anyio_backend) -> AsyncGenerator[AsyncConnection, None]:
+async def connection(
+    anyio_backend: str,  # noqa: ARG001 - activates AnyIO's async-fixture runner
+) -> AsyncGenerator[AsyncConnection, None]:
     async with engine.connect() as connection:
         yield connection
 
 
-@pytest.fixture()
+@pytest.fixture(scope="session", autouse=True)
+async def create_metadata(
+    connection: AsyncConnection,
+) -> AsyncGenerator[None, None]:
+    await connection.run_sync(Base.metadata.create_all)
+    await connection.run_sync(CustomPKBase.metadata.create_all)
+    await connection.commit()
+
+    yield
+
+    await connection.run_sync(CustomPKBase.metadata.drop_all)
+    await connection.run_sync(Base.metadata.drop_all)
+    await connection.commit()
+
+
+@pytest.fixture
 async def transaction(
     connection: AsyncConnection,
 ) -> AsyncGenerator[AsyncTransaction, None]:
@@ -39,7 +49,7 @@ async def persistent_session() -> AsyncGenerator[AsyncSession, None]:
         yield session
 
 
-@pytest.fixture()
+@pytest.fixture
 async def session(
     connection: AsyncConnection, transaction: AsyncTransaction
 ) -> AsyncGenerator[AsyncSession, None]:

tests/models.py

@@ -2,7 +2,17 @@
 from uuid import UUID, uuid4
 
 from pydantic import BaseModel
-from sqlalchemy import Column, DateTime, ForeignKey, Table, UniqueConstraint, func
+from sqlalchemy import (
+    Column,
+    DateTime,
+    ForeignKey,
+    Index,
+    Table,
+    UniqueConstraint,
+    column,
+    func,
+    text,
+)
 from sqlalchemy.orm import (
     DeclarativeBase,
     Mapped,
@@ -112,3 +122,55 @@ class ItemSchema(BaseModel):
 
 
 item_manager = ModelManager[Item, ItemSchema, ItemSchema](Item)
+
+
+class PartialUniqueText(Base):
+    """Partial unique index via text() postgresql_where."""
+
+    group_id: Mapped[UUID]
+    kind: Mapped[str]
+
+    __table_args__ = (
+        Index(
+            "uix_partial_unique_text_one_main_per_group",
+            "group_id",
+            unique=True,
+            postgresql_where=text("kind = 'main'"),
+        ),
+    )
+
+
+class PartialUniqueTextSchema(BaseModel):
+    group_id: UUID
+    kind: str
+
+
+partial_unique_text_manager = ModelManager[
+    PartialUniqueText, PartialUniqueTextSchema, PartialUniqueTextSchema
+](PartialUniqueText)
+
+
+class PartialUniqueExpr(Base):
+    """Partial unique index via Column expression postgresql_where."""
+
+    group_id: Mapped[UUID]
+    kind: Mapped[str]
+
+    __table_args__ = (
+        Index(
+            "uix_partial_unique_expr_one_main_per_group",
+            "group_id",
+            unique=True,
+            postgresql_where=(column("kind") == "main"),
+        ),
+    )
+
+
+class PartialUniqueExprSchema(BaseModel):
+    group_id: UUID
+    kind: str
+
+
+partial_unique_expr_manager = ModelManager[
+    PartialUniqueExpr, PartialUniqueExprSchema, PartialUniqueExprSchema
+](PartialUniqueExpr)