Merge pull request #387 from d3vyce/386-pagination-over-a-to-many-join-duplicates-drops-and-truncates-results

fix: pagination over a to-many join duplicates, drops and truncates results

docs/module/db.md

@@ -54,7 +54,7 @@ db = Database(
 
 ## Committing before the response
 
-[`db.install(app)`](../reference/db.md#fastapi_toolsets.db.Database) adds a middleware that commits the request's session when the response starts, after the endpoint returns and before the body is sent. With the middleware installed, the dependency does not commit again.
+[`db.install(app)`](../reference/db.md#fastapi_toolsets.db.Database) adds a middleware that commits the request's session when the response starts, after the endpoint returns and before the body is sent. The dependency commits only if the middleware did not: when a function-scoped dependency unwinds before the response, or when the response never passes through the middleware. Either way the request is committed exactly once.
 
 The request is committed as a single transaction:
 

pyproject.toml

@@ -101,7 +101,7 @@ exclude = ["*.md"]
 extend-select = ["E712"]
 
 [tool.ruff.lint.flake8-bugbear]
-extend-immutable-calls = ["fastapi.Depends"]
+extend-immutable-calls = ["fastapi.Depends", "fastapi.Security"]
 
 [tool.ruff.lint.per-file-ignores]
 "tests/**" = ["RUF012", "RUF059", "SIM117", "DTZ001", "S110", "BLE001"]

src/fastapi_toolsets/crud/factory.py

@@ -14,12 +14,25 @@ from typing import Any, ClassVar, Generic, Literal, Self, TypeAlias, cast, overl
 
 from fastapi import Query
 from pydantic import BaseModel
-from sqlalchemy import Date, DateTime, Float, Integer, Numeric, Uuid, and_, func, select
+from sqlalchemy import (
+    Date,
+    DateTime,
+    Float,
+    Integer,
+    Numeric,
+    Uuid,
+    and_,
+    func,
+    select,
+    tuple_,
+)
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import DeclarativeBase, QueryableAttribute, selectinload
+from sqlalchemy.sql import operators
 from sqlalchemy.sql.base import ExecutableOption
+from sqlalchemy.sql.elements import UnaryExpression
 from sqlalchemy.sql.roles import WhereHavingRole
 
 from ..db import transaction
@@ -128,6 +141,32 @@ def _apply_joins(q: Any, joins: JoinType | None, outer_join: bool) -> Any:
     return q
 
 
+def _fans_out(
+    search_joins: Sequence[Any] | None, order_joins: Sequence[Any] | None
+) -> bool:
+    """True if any relationship join yields a collection."""
+    return any(
+        rel.property.uselist for rel in (*(search_joins or ()), *(order_joins or ()))
+    )
+
+
+def _grouped_order(clause: Any, table: Any) -> Any:
+    """Recast an order clause for a query grouped by the entity's key."""
+    inner = clause.element if isinstance(clause, UnaryExpression) else clause
+    expr = inner.__clause_element__() if hasattr(inner, "__clause_element__") else inner
+    tables = {
+        t
+        for c in getattr(expr, "base_columns", ())
+        if (t := getattr(c, "table", None)) is not None
+    }
+    if tables and tables <= {table}:
+        return clause
+    agg = func.min(inner)
+    if isinstance(clause, UnaryExpression) and clause.modifier is operators.desc_op:
+        return agg.desc()
+    return agg.asc()
+
+
 class AsyncCrud(Generic[ModelType]):
     """Generic async CRUD operations for SQLAlchemy models.
 
@@ -163,6 +202,64 @@ class AsyncCrud(Generic[ModelType]):
             ):
                 cls.searchable_fields = [pk_col, *raw_fields]
 
+    @classmethod
+    def _pk_attrs(cls: type[Self]) -> list[QueryableAttribute[Any]]:
+        """The model's primary key columns as instrumented attributes."""
+        return [
+            getattr(cls.model, cast(str, col.key))
+            for col in cls.model.__mapper__.primary_key
+        ]
+
+    @classmethod
+    async def _page_entities(
+        cls: type[Self],
+        session: AsyncSession,
+        q: Any,
+        *,
+        order_clauses: Sequence[Any],
+        limit: int,
+        offset: int | None = None,
+        load_options: Sequence[ExecutableOption] | None = None,
+        with_for_update: _ForUpdateMode = False,
+    ) -> list[ModelType]:
+        """Return up to *limit* entities, paging over distinct primary keys."""
+        pk_attrs = cls._pk_attrs()
+        table = cls.model.__table__
+        # Fall back to the primary key so the page boundary is deterministic.
+        grouped = [_grouped_order(c, table) for c in order_clauses] or [pk_attrs[0]]
+        id_q = (
+            q.order_by(None)
+            .with_only_columns(*pk_attrs)
+            .group_by(*pk_attrs)
+            .order_by(*grouped)
+            .limit(limit)
+        )
+        if offset:
+            id_q = id_q.offset(offset)
+        rows = (await session.execute(id_q)).all()
+        ids = [row[0] if len(pk_attrs) == 1 else tuple(row) for row in rows]
+        if not ids:
+            return []
+
+        where = (
+            pk_attrs[0].in_(ids) if len(pk_attrs) == 1 else tuple_(*pk_attrs).in_(ids)
+        )
+        item_q = select(cls.model).where(where)
+        if resolved := cls._resolve_load_options(load_options):
+            item_q = item_q.options(*resolved)
+        item_q = _apply_for_update(item_q, with_for_update)
+        found = (await session.execute(item_q)).unique().scalars().all()
+
+        rank = {pk: n for n, pk in enumerate(ids)}
+
+        def _key(obj: Any) -> Any:
+            values = tuple(getattr(obj, a.key) for a in pk_attrs)
+            return values[0] if len(pk_attrs) == 1 else values
+
+        return cast(
+            list[ModelType], sorted(found, key=lambda o: rank.get(_key(o), len(ids)))
+        )
+
     @classmethod
     def _resolve_load_options(
         cls, load_options: Sequence[ExecutableOption] | None
@@ -1023,9 +1120,21 @@ class AsyncCrud(Generic[ModelType]):
             q = q.where(and_(*filters))
         if resolved := cls._resolve_load_options(load_options):
             q = q.options(*resolved)
-        q = _apply_for_update(q, with_for_update)
         if order_by is not None:
             q = q.order_by(order_by)
+
+        if limit is not None and joins:
+            return await cls._page_entities(
+                session,
+                q,
+                order_clauses=[] if order_by is None else [order_by],
+                limit=limit,
+                offset=offset,
+                load_options=load_options,
+                with_for_update=with_for_update,
+            )
+
+        q = _apply_for_update(q, with_for_update)
         if offset is not None:
             q = q.offset(offset)
         if limit is not None:
@@ -1374,17 +1483,31 @@ class AsyncCrud(Generic[ModelType]):
             q = q.where(and_(*filters))
         if resolved := cls._resolve_load_options(load_options):
             q = q.options(*resolved)
-        if order_by is not None:
-            q = q.order_by(order_by)
+        order_clauses: list[Any] = [] if order_by is None else [order_by]
+        q = q.order_by(*order_clauses)
+
+        fetch_limit = items_per_page if include_total else items_per_page + 1
+        total_count: int | None = None
+        # A to-many join repeats each entity, so LIMIT would slice joined rows
+        # and `.unique()` would shrink the page after the fact.
+        if _fans_out(search_joins, order_joins):
+            raw_items = await cls._page_entities(
+                session,
+                q,
+                order_clauses=order_clauses,
+                limit=fetch_limit,
+                offset=offset,
+                load_options=load_options,
+            )
+        else:
+            result = await session.execute(q.offset(offset).limit(fetch_limit))
+            raw_items = cast(list[ModelType], result.unique().scalars().all())
+        fetched = len(raw_items)
+        raw_items = raw_items[:items_per_page]
 
         if include_total:
-            q = q.offset(offset).limit(items_per_page)
-            result = await session.execute(q)
-            raw_items = cast(list[ModelType], result.unique().scalars().all())
-
             # Count query (with same joins and filters)
-            pk_col = cls.model.__mapper__.primary_key[0]
-            count_q = select(func.count(func.distinct(getattr(cls.model, pk_col.name))))
+            count_q = select(func.count(func.distinct(cls._pk_attrs()[0])))
             count_q = count_q.select_from(cls.model)
 
             # Apply explicit joins to count query
@@ -1397,16 +1520,11 @@ class AsyncCrud(Generic[ModelType]):
                 count_q = count_q.where(and_(*filters))
 
             count_result = await session.execute(count_q)
-            total_count: int = count_result.scalar_one()
+            total_count = count_result.scalar_one()
             has_more = page * items_per_page < total_count
         else:
-            # Fetch one extra row to detect if a next page exists without COUNT
-            q = q.offset(offset).limit(items_per_page + 1)
-            result = await session.execute(q)
-            raw_items = cast(list[ModelType], result.unique().scalars().all())
-            has_more = len(raw_items) > items_per_page
-            raw_items = raw_items[:items_per_page]
-            total_count = None
+            # One extra row was fetched to detect a next page without COUNT
+            has_more = fetched > items_per_page
 
         items: list[Any] = [schema.model_validate(item) for item in raw_items]
 
@@ -1543,18 +1661,30 @@ class AsyncCrud(Generic[ModelType]):
             q = q.options(*resolved)
 
         # Cursor column is always the primary sort; reverse direction for prev traversal
-        if direction is _CursorDirection.PREV:
-            q = q.order_by(cursor_column.desc())
-        else:
-            q = q.order_by(cursor_column)
+        cursor_clause = (
+            cursor_column.desc()
+            if direction is _CursorDirection.PREV
+            else cursor_column
+        )
+        order_clauses: list[Any] = [cursor_clause]
         if order_by is not None:
-            q = q.order_by(order_by)
+            order_clauses.append(order_by)
+        q = q.order_by(*order_clauses)
 
-        # Fetch one extra to detect whether another page exists in this direction
-        q = q.limit(items_per_page + 1)
-        result = await session.execute(q)
+        # One extra row detects whether another page exists in this direction.
+        # Under a to-many join that extra row may be a duplicate of one already
+        # on the page, which reads as "no next page" and ends traversal early.
+        if _fans_out(search_joins, order_joins):
+            raw_items = await cls._page_entities(
+                session,
+                q,
+                order_clauses=order_clauses,
+                limit=items_per_page + 1,
+                load_options=load_options,
+            )
+        else:
+            result = await session.execute(q.limit(items_per_page + 1))
             raw_items = cast(list[ModelType], result.unique().scalars().all())
-
         has_more = len(raw_items) > items_per_page
         items_page = raw_items[:items_per_page]
 

src/fastapi_toolsets/crud/search.py

@@ -160,9 +160,11 @@ def build_search_filters(
             column = field
 
         # Build the filter (cast to String only when needed, to preserve
-        # pg_trgm GIN index usability on already-String columns)
+        # pg_trgm GIN index usability on already-String columns).
         column_as_string = (
-            column if isinstance(column.type, String) else column.cast(String)
+            column
+            if isinstance(column.type, String) and not isinstance(column.type, Enum)
+            else column.cast(String)
         )
         if config.case_sensitive:
             filters.append(column_as_string.like(f"%{query}%"))

src/fastapi_toolsets/db/core.py

@@ -66,10 +66,8 @@ class _CommitOnResponseMiddleware:
 
         async def send_wrapper(message: Message) -> None:
             if message["type"] == "http.response.start":
-                # ``scope["state"]`` is the same dict ``request.state`` writes
-                # to, so this is the session stashed by the dependency.
                 state = scope.get("state")
-                session = state.get(self.state_attr) if state else None
+                session = state.pop(self.state_attr, None) if state else None
                 if session is not None and session.in_transaction():
                     await session.commit()
             await send(message)
@@ -158,7 +156,6 @@ class Database:
         # Private, per-instance state attribute; cannot collide with another
         # Database or be mismatched against the middleware.
         self._state_attr = f"_ft_db_session_{id(self):x}"
-        self._middleware_installed = False
         self._disposed = False
 
     async def _dispose(self) -> None:
@@ -206,7 +203,6 @@ class Database:
             ```
         """
         app.add_middleware(_CommitOnResponseMiddleware, state_attr=self._state_attr)
-        self._middleware_installed = True
 
         inner_lifespan = app.router.lifespan_context
 
@@ -243,10 +239,17 @@ class Database:
                 return await UserCrud.get(session, [User.id == user_id])
             ```
         """
+        borrowed = getattr(request.state, self._state_attr, None)
+        if borrowed is not None:
+            yield borrowed
+            return
         async with self._open() as session:
             setattr(request.state, self._state_attr, session)
             yield session
-            if not self._middleware_installed and session.in_transaction():
+            if (
+                getattr(request.state, self._state_attr, None) is session
+                and session.in_transaction()
+            ):
                 await session.commit()
 
     @asynccontextmanager

src/fastapi_toolsets/models/watched.py

@@ -8,6 +8,7 @@ from typing import Any
 from sqlalchemy import event, select, tuple_
 from sqlalchemy import inspect as sa_inspect
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
 from sqlalchemy.orm.attributes import set_committed_value as _sa_set_committed_value
 
 from ..logger import get_logger
@@ -189,17 +190,44 @@ async def _invoke_callback(
         await result
 
 
+def _loaded_relationships(obj: Any) -> set[str]:
+    """Relationship keys currently loaded on *obj*."""
+    state = sa_inspect(obj)
+    unloaded = state.unloaded
+    return {
+        rel.key
+        for rel in state.mapper.relationships
+        if rel.key not in unloaded and rel.lazy not in ("dynamic", "write_only")
+    }
+
+
+def _snapshot_loaded_relationships(session: Any) -> dict[int, set[str]]:
+    """Record loaded relationships for the tracked objects, keyed by ``id``."""
+    objs = list(session.info.get(_SESSION_CREATES, []))
+    objs += [obj for obj, _ in session.info.get(_SESSION_UPDATES, {}).values()]
+    return {id(obj): _loaded_relationships(obj) for obj in objs}
+
+
 async def _batch_reload(
-    session: AsyncSession, model: type, pk_tuples: list[tuple[Any, ...]]
+    session: AsyncSession,
+    model: type,
+    objs: list[Any],
+    preloaded: dict[int, set[str]],
 ) -> None:
-    """Re-populate all rows of *model* identified by *pk_tuples* in one round trip."""
+    """Re-populate all rows of *model* in one round trip."""
     pk_cols = sa_inspect(model, raiseerr=True).primary_key
+    pk_tuples = [sa_inspect(obj).key[1] for obj in objs]
     where = (
         pk_cols[0].in_([pk[0] for pk in pk_tuples])
         if len(pk_cols) == 1
         else tuple_(*pk_cols).in_(pk_tuples)
     )
     q = select(model).where(where).execution_options(populate_existing=True)
+    loaded: set[str] = set()
+    for obj in objs:
+        loaded |= preloaded.get(id(obj), set())
+    if loaded:
+        q = q.options(*(selectinload(getattr(model, key)) for key in loaded))
     await session.execute(q)
 
 
@@ -207,6 +235,7 @@ class EventSession(AsyncSession):
     """AsyncSession subclass that dispatches lifecycle callbacks after commit."""
 
     async def commit(self) -> None:
+        preloaded = _snapshot_loaded_relationships(self)
         await super().commit()
 
         creates: list[Any] = self.info.pop(_SESSION_CREATES, [])
@@ -249,25 +278,25 @@ class EventSession(AsyncSession):
         # session.get() per object.
         create_items: list[Any] = []
         update_items: list[tuple[Any, dict[str, dict[str, Any]]]] = []
-        pk_by_type: dict[type, list[tuple[Any, ...]]] = {}
+        objs_by_type: dict[type, list[Any]] = {}
 
         for obj in creates:
             state = sa_inspect(obj, raiseerr=False)
             if state is None or state.detached or state.transient:  # pragma: no cover
                 continue
             create_items.append(obj)
-            pk_by_type.setdefault(type(obj), []).append(state.key[1])
+            objs_by_type.setdefault(type(obj), []).append(obj)
 
         for obj, changes in field_changes.values():
             state = sa_inspect(obj, raiseerr=False)
             if state is None or state.detached or state.transient:  # pragma: no cover
                 continue
             update_items.append((obj, changes))
-            pk_by_type.setdefault(type(obj), []).append(state.key[1])
+            objs_by_type.setdefault(type(obj), []).append(obj)
 
-        for model, pk_tuples in pk_by_type.items():
+        for model, objs in objs_by_type.items():
             try:
-                await _batch_reload(self, model, pk_tuples)
+                await _batch_reload(self, model, objs, preloaded)
             except Exception as exc:
                 _logger.error(_CALLBACK_ERROR_MSG, exc_info=exc)
 

tests/test_crud_search.py

@@ -5,6 +5,7 @@ import uuid
 
 import pytest
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
 from sqlalchemy.sql.elements import ColumnElement, UnaryExpression
 
 from fastapi_toolsets.crud import (
@@ -16,7 +17,7 @@ from fastapi_toolsets.crud import (
     get_searchable_fields,
 )
 from fastapi_toolsets.exceptions import InvalidOrderFieldError
-from fastapi_toolsets.schemas import OffsetPagination, PaginationType
+from fastapi_toolsets.schemas import OffsetPagination, PaginationType, PydanticBase
 
 from .conftest import (
     Article,
@@ -29,15 +30,19 @@ from .conftest import (
     OrderCrud,
     OrderRead,
     OrderStatus,
+    Post,
+    PostCrud,
     Role,
     RoleCreate,
     RoleCrud,
     RoleCursorCrud,
     RoleRead,
+    Tag,
     User,
     UserCreate,
     UserCrud,
     UserRead,
+    post_tags,
 )
 
 
@@ -388,6 +393,295 @@ class TestBuildSearchFilters:
 
         assert "CAST" in str(filters[0])
 
+    def test_casts_enum_column(self):
+        """Enum subclasses String but maps to a native DB enum, which has no ILIKE."""
+        from fastapi_toolsets.crud.search import build_search_filters
+
+        filters, _ = build_search_filters(Order, "PEND", search_fields=[Order.status])
+
+        assert "CAST" in str(filters[0])
+
+
+class _PostTitle(PydanticBase):
+    """Minimal read schema for the to-many pagination tests."""
+
+    id: uuid.UUID
+    title: str
+
+
+class _TagName(PydanticBase):
+    name: str
+
+
+class _PostWithTags(PydanticBase):
+    """Serialising `tags` fails unless the relation was eager-loaded."""
+
+    title: str
+    tags: list[_TagName]
+
+
+PostTagSearchCrud = CrudFactory(
+    Post,
+    searchable_fields=[Post.title, (Post.tags, Tag.name)],
+    cursor_column=Post.id,
+)
+
+_POST_COUNT = 10
+
+
+async def _seed_posts_with_tags(session) -> None:
+    """10 posts, each with 3 tags whose names all match the search term."""
+    author = await UserCrud.create(
+        session, UserCreate(username="fanout", email="fanout@test.com")
+    )
+    for i in range(_POST_COUNT):
+        tags = [Tag(name=f"shared-{i}-{j}") for j in range(3)]
+        session.add_all(tags)
+        session.add(Post(title=f"post{i:02d}", author_id=author.id, tags=tags))
+    await session.flush()
+
+
+class TestPaginateToManyJoin:
+    """Searching a to-many relationship must not truncate or duplicate pages."""
+
+    @pytest.mark.anyio
+    async def test_offset_pages_are_full_and_complete(self, db_session: AsyncSession):
+        """Every page is full and every post is returned exactly once."""
+        await _seed_posts_with_tags(db_session)
+
+        seen: list[str] = []
+        for page in (1, 2):
+            result = await PostTagSearchCrud.offset_paginate(
+                db_session,
+                page=page,
+                items_per_page=5,
+                search="shared",
+                schema=_PostTitle,
+            )
+            assert result.pagination.total_count == _POST_COUNT
+            assert len(result.data) == 5, f"page {page} came back short"
+            seen += [p.title for p in result.data]
+
+        assert len(set(seen)) == _POST_COUNT, "posts duplicated or missing"
+
+    @pytest.mark.anyio
+    async def test_offset_without_total_reports_has_more(
+        self, db_session: AsyncSession
+    ):
+        """``has_more`` counts entities, not joined rows."""
+        await _seed_posts_with_tags(db_session)
+
+        first = await PostTagSearchCrud.offset_paginate(
+            db_session,
+            page=1,
+            items_per_page=5,
+            search="shared",
+            include_total=False,
+            schema=_PostTitle,
+        )
+        assert len(first.data) == 5
+        assert first.pagination.has_more is True
+
+        last = await PostTagSearchCrud.offset_paginate(
+            db_session,
+            page=2,
+            items_per_page=5,
+            search="shared",
+            include_total=False,
+            schema=_PostTitle,
+        )
+        assert len(last.data) == 5
+        assert last.pagination.has_more is False
+
+    @pytest.mark.anyio
+    async def test_cursor_traverses_every_row(self, db_session: AsyncSession):
+        """Cursor traversal must not stop early."""
+        await _seed_posts_with_tags(db_session)
+
+        seen: list[str] = []
+        cursor: str | None = None
+        for _ in range(_POST_COUNT):
+            result = await PostTagSearchCrud.cursor_paginate(
+                db_session,
+                cursor=cursor,
+                items_per_page=5,
+                search="shared",
+                schema=_PostTitle,
+            )
+            seen += [p.title for p in result.data]
+            cursor = result.pagination.next_cursor
+            if cursor is None:
+                break
+
+        assert len(seen) == _POST_COUNT, "traversal stopped early or repeated rows"
+        assert len(set(seen)) == _POST_COUNT
+
+    @pytest.mark.anyio
+    async def test_orders_by_a_to_many_column(self, db_session: AsyncSession):
+        """Ordering by a related column has to be collapsed to an aggregate."""
+        await _seed_posts_with_tags(db_session)
+
+        result = await PostTagSearchCrud.offset_paginate(
+            db_session,
+            page=1,
+            items_per_page=5,
+            search="shared",
+            order_by=Tag.name.asc(),
+            order_joins=[Post.tags],
+            schema=_PostTitle,
+        )
+
+        assert len(result.data) == 5
+        assert result.pagination.total_count == _POST_COUNT
+
+    @pytest.mark.anyio
+    async def test_orders_by_a_bare_to_many_column(self, db_session: AsyncSession):
+        """An order clause with no explicit direction still needs aggregating."""
+        await _seed_posts_with_tags(db_session)
+
+        result = await PostTagSearchCrud.offset_paginate(
+            db_session,
+            page=1,
+            items_per_page=5,
+            search="shared",
+            order_by=Tag.name,
+            order_joins=[Post.tags],
+            schema=_PostTitle,
+        )
+
+        assert len(result.data) == 5
+
+    @pytest.mark.anyio
+    async def test_page_past_the_end_is_empty(self, db_session: AsyncSession):
+        """No keys on the page means no second query and an empty result."""
+        await _seed_posts_with_tags(db_session)
+
+        result = await PostTagSearchCrud.offset_paginate(
+            db_session, page=99, items_per_page=5, search="shared", schema=_PostTitle
+        )
+
+        assert result.data == []
+        assert result.pagination.total_count == _POST_COUNT
+
+    @pytest.mark.anyio
+    async def test_load_options_apply_on_the_fan_out_path(
+        self, db_session: AsyncSession
+    ):
+        """The entity query still honours loader options."""
+        await _seed_posts_with_tags(db_session)
+
+        result = await PostTagSearchCrud.offset_paginate(
+            db_session,
+            page=1,
+            items_per_page=5,
+            search="shared",
+            load_options=[selectinload(Post.tags)],
+            schema=_PostWithTags,
+        )
+
+        assert len(result.data) == 5
+        assert all(len(p.tags) == 3 for p in result.data)
+
+    @pytest.mark.anyio
+    async def test_get_multi_is_not_truncated_by_a_to_many_join(
+        self, db_session: AsyncSession
+    ):
+        """`get_multi` takes a raw join, so cardinality cannot be inspected."""
+        await _seed_posts_with_tags(db_session)
+
+        rows = await PostCrud.get_multi(
+            db_session,
+            joins=[(post_tags, post_tags.c.post_id == Post.id)],
+            outer_join=True,
+            limit=5,
+        )
+
+        assert len(rows) == 5
+        assert len({r.id for r in rows}) == 5
+
+    def test_grouped_order_only_aggregates_foreign_columns(self):
+        """A base-table column is left alone; anything else collapses to min()."""
+        from fastapi_toolsets.crud.factory import _grouped_order
+
+        table = Post.__table__
+
+        assert "min" not in str(_grouped_order(Post.title.desc(), table)).lower()
+
+        asc_on_join = str(_grouped_order(Tag.name.asc(), table))
+        desc_on_join = str(_grouped_order(Tag.name.desc(), table))
+        assert "min" in asc_on_join.lower() and asc_on_join.endswith("ASC")
+        assert "min" in desc_on_join.lower() and desc_on_join.endswith("DESC")
+
+    def test_to_one_join_does_not_take_the_fan_out_path(self):
+        """A to-one join keeps the single-query path."""
+        from fastapi_toolsets.crud.factory import _fans_out
+
+        assert _fans_out([User.role], None) is False
+        assert _fans_out([Post.tags], None) is True
+        assert _fans_out(None, [Post.tags]) is True
+
+
+class TestSearchEnumColumn:
+    """Searching an enum column must reach the database, not just build SQL."""
+
+    @pytest.mark.anyio
+    async def test_search_int_backed_enum(self, db_session: AsyncSession):
+        """Enum(int, Enum) stores names, so the cast makes 'PEND' match PENDING."""
+        await OrderCrud.create(
+            db_session, OrderCreate(name="a", status=OrderStatus.PENDING)
+        )
+        await OrderCrud.create(
+            db_session, OrderCreate(name="b", status=OrderStatus.SHIPPED)
+        )
+
+        result = await OrderCrud.offset_paginate(
+            db_session,
+            search="PEND",
+            search_fields=[Order.status],
+            schema=OrderRead,
+        )
+
+        assert result.pagination.total_count == 1
+        assert result.data[0].status is OrderStatus.PENDING
+
+    @pytest.mark.anyio
+    async def test_search_str_backed_enum(self, db_session: AsyncSession):
+        """Same for Enum(str, Enum) — still a native DB enum, still needs the cast."""
+        await OrderCrud.create(
+            db_session,
+            OrderCreate(name="a", status=OrderStatus.PENDING, color=Color.BLUE),
+        )
+        await OrderCrud.create(
+            db_session,
+            OrderCreate(name="b", status=OrderStatus.PENDING, color=Color.RED),
+        )
+
+        result = await OrderCrud.offset_paginate(
+            db_session,
+            search="BLU",
+            search_fields=[Order.color],
+            schema=OrderRead,
+        )
+
+        assert result.pagination.total_count == 1
+        assert result.data[0].color is Color.BLUE
+
+    @pytest.mark.anyio
+    async def test_search_mixed_enum_and_string_columns(self, db_session: AsyncSession):
+        """An enum column alongside a plain String column (the get_searchable_fields shape)."""
+        await OrderCrud.create(
+            db_session, OrderCreate(name="widget", status=OrderStatus.SHIPPED)
+        )
+
+        result = await OrderCrud.offset_paginate(
+            db_session,
+            search="widget",
+            search_fields=[Order.name, Order.status, Order.color],
+            schema=OrderRead,
+        )
+
+        assert result.pagination.total_count == 1
+
 
 class TestSearchConfig:
     """Tests for SearchConfig options."""

tests/test_db.py

@@ -6,7 +6,7 @@ from contextlib import asynccontextmanager
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from fastapi import Depends, FastAPI
+from fastapi import Depends, FastAPI, Security
 from fastapi.responses import StreamingResponse
 from httpx import ASGITransport, AsyncClient
 from pydantic import PostgresDsn
@@ -287,23 +287,37 @@ class TestDatabaseDependency:
             break
 
     @pytest.mark.anyio
-    async def test_skips_commit_when_middleware_installed(self, engine, session_maker):
-        """With ``install()``, the dependency must NOT commit — the middleware owns it.
+    async def test_second_resolution_borrows_session(self, engine):
+        """A second ``Depends(db)`` in one request reuses the stashed session."""
+        db = Database(engine=engine)
+        request = _make_request()
 
-        Here no middleware actually runs (we call the dependency directly), so the
-        open transaction is rolled back on session close and nothing persists.
-        """
+        owner_gen = db(request)
+        owner = await anext(owner_gen)
+        borrower_gen = db(request)
+        assert await anext(borrower_gen) is owner
+
+        with pytest.raises(StopAsyncIteration):  # teardown runs borrower-first
+            await anext(borrower_gen)
+        assert owner.in_transaction()  # the borrower must not close what it borrowed
+
+        with pytest.raises(StopAsyncIteration):
+            await anext(owner_gen)
+
+    @pytest.mark.anyio
+    async def test_commits_when_middleware_did_not_run(self, engine, session_maker):
+        """``install()`` is per-``Database``, but the commit is per-request."""
         db = Database(engine=engine)
         db.install(FastAPI())
 
         async for session in db(_make_request()):
-            role = Role(name="mw_owns_commit")
+            role = Role(name="mw_never_ran")
             session.add(role)
             await session.flush()
 
         async with session_maker() as verify:
-            result = await RoleCrud.first(verify, [Role.name == "mw_owns_commit"])
-            assert result is None
+            result = await RoleCrud.first(verify, [Role.name == "mw_never_ran"])
+            assert result is not None
 
 
 class TestDatabaseSession:
@@ -1523,6 +1537,55 @@ def _build_app(db: Database) -> FastAPI:
         await session.commit()
         return {"id": str(role.id), "name": role.name}
 
+    async def _scoped_writer(
+        body: RoleCreate, session: AsyncSession = Security(db, scopes=["roles:write"])
+    ) -> int:
+        # Security scopes give this a different dependency cache key than the
+        # endpoint's plain ``Depends(db)``. Without borrowing it opens a second
+        # session, and whichever one the middleware does not hold is discarded.
+        await RoleCrud.create(session, RoleCreate(name=f"{body.name}_sub"))
+        return id(session)
+
+    @app.post("/roles-two-cache-keys")
+    async def create_via_two_cache_keys(
+        body: RoleCreate,
+        sub_session_id: int = Depends(_scoped_writer),
+        session: AsyncSession = Depends(db),
+    ) -> dict:
+        await RoleCrud.create(session, body)
+        return {"same_session": sub_session_id == id(session)}
+
+    async def _fn_writer(
+        body: RoleCreate, session: AsyncSession = Depends(db, scope="function")
+    ) -> None:
+        # ``scope="function"`` unwinds before the response is sent, taking the
+        # session with it — so the commit cannot be left to the middleware.
+        await RoleCrud.create(session, RoleCreate(name=f"{body.name}_fn"))
+
+    @app.post("/roles-function-scope")
+    async def create_with_function_scope(
+        body: RoleCreate,
+        boom: bool = False,
+        _: None = Depends(_fn_writer),
+        session: AsyncSession = Depends(db),
+    ) -> dict:
+        await RoleCrud.create(session, body)
+        if boom:
+            raise RuntimeError("boom after write")
+        return {"ok": True}
+
+    @app.post("/roles-function-scope-borrower")
+    async def function_scope_borrows(
+        body: RoleCreate,
+        session: AsyncSession = Depends(db),
+        _: None = Depends(_fn_writer),
+    ) -> dict:
+        # Flipped order: the request-scoped dependency owns the session and the
+        # function-scoped one borrows it. The borrower unwinds early but must not
+        # commit or close — the commit still belongs to the middleware.
+        await RoleCrud.create(session, body)
+        return {"ok": True}
+
     @app.get("/roles-stream/{name}")
     async def stream_role(
         name: str, session: AsyncSession = Depends(db)
@@ -1619,6 +1682,62 @@ class TestCommitIntegration:
         # The write made before the stream began is durably committed.
         assert await _row_exists(session_maker, "streamed_role")
 
+    @pytest.mark.anyio
+    async def test_two_cache_keys_share_one_session(self, engine, session_maker):
+        """Two resolutions of ``Depends(db)`` in one request must share a session."""
+        app = _build_app(Database(engine=engine))
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            resp = await client.post("/roles-two-cache-keys", json={"name": "two_keys"})
+
+        assert resp.status_code == 200
+        assert resp.json()["same_session"] is True
+        assert await _row_exists(session_maker, "two_keys")
+        assert await _row_exists(session_maker, "two_keys_sub")
+
+    @pytest.mark.anyio
+    async def test_function_scope_commits_before_response(self, engine, session_maker):
+        """``scope="function"`` unwinds before response-start, so the dependency
+        commits on its way out instead of leaving it to the middleware."""
+        app = _build_app(Database(engine=engine))
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            resp = await client.post("/roles-function-scope", json={"name": "fn_scope"})
+
+        assert resp.status_code == 200
+        assert await _row_exists(session_maker, "fn_scope")
+        assert await _row_exists(session_maker, "fn_scope_fn")
+
+    @pytest.mark.anyio
+    async def test_function_scope_borrower_leaves_commit_to_middleware(
+        self, engine, session_maker
+    ):
+        """A function-scoped *borrower* unwinds early but owns nothing."""
+        app = _build_app(Database(engine=engine))
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            resp = await client.post(
+                "/roles-function-scope-borrower", json={"name": "fn_borrow"}
+            )
+
+        assert resp.status_code == 200
+        assert await _row_exists(session_maker, "fn_borrow")
+        assert await _row_exists(session_maker, "fn_borrow_fn")
+
+    @pytest.mark.anyio
+    async def test_function_scope_error_rolls_back(self, engine, session_maker):
+        """The early commit must still not fire when the request fails."""
+        app = _build_app(Database(engine=engine))
+        transport = ASGITransport(app=app, raise_app_exceptions=False)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            resp = await client.post(
+                "/roles-function-scope?boom=true", json={"name": "fn_ghost"}
+            )
+
+        assert resp.status_code == 500
+        assert not await _row_exists(session_maker, "fn_ghost")
+        assert not await _row_exists(session_maker, "fn_ghost_fn")
+
     @pytest.mark.anyio
     async def test_multi_write_atomicity(self, engine, session_maker):
         """When the 2nd write fails, the 1st must roll back too (one txn)."""

tests/test_models.py

@@ -6,9 +6,16 @@ from types import SimpleNamespace
 from unittest.mock import patch
 
 import pytest
-from sqlalchemy import String
+from sqlalchemy import ForeignKey, String, select
+from sqlalchemy import inspect as sa_inspect
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+from sqlalchemy.orm import (
+    DeclarativeBase,
+    Mapped,
+    mapped_column,
+    relationship,
+    selectinload,
+)
 
 import fastapi_toolsets.models.watched as _watched_module
 from fastapi_toolsets.models import (
@@ -107,6 +114,27 @@ async def _watched_on_update(obj, event_type, changes):
     _test_events.append({"event": "update", "obj_id": obj.id, "changes": changes})
 
 
+class RelTarget(MixinBase, UUIDMixin):
+    __tablename__ = "mixin_rel_targets"
+
+    name: Mapped[str] = mapped_column(String(50))
+
+
+class RelOwner(MixinBase, UUIDMixin):
+    """Watched model with a relationship, to check eager loads survive commit."""
+
+    __tablename__ = "mixin_rel_owners"
+
+    title: Mapped[str] = mapped_column(String(50))
+    target_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("mixin_rel_targets.id"))
+    target: Mapped[RelTarget] = relationship()
+
+
+@listens_for(RelOwner, [ModelEvent.CREATE, ModelEvent.UPDATE])
+async def _rel_owner_handler(obj, event_type, changes):
+    _test_events.append({"event": event_type.value, "obj_id": obj.id})
+
+
 class WatchAllModel(MixinBase, UUIDMixin):
     """Model without __watched_fields__ — watches all mapped fields by default."""
 
@@ -355,6 +383,62 @@ async def mixin_session_maker():
         await engine.dispose()
 
 
+class TestEventSessionPreservesEagerLoads:
+    """EventSession.commit() must not discard relations an eager load populated."""
+
+    async def _seed_eager(self, session):
+        target = RelTarget(name="t")
+        session.add(target)
+        await session.flush()
+        owner = RelOwner(title="o", target_id=target.id)
+        session.add(owner)
+        await session.flush()
+        loaded = (
+            await session.execute(
+                select(RelOwner)
+                .where(RelOwner.id == owner.id)
+                .options(selectinload(RelOwner.target))
+            )
+        ).scalar_one()
+        assert "target" not in sa_inspect(loaded).unloaded
+        return loaded
+
+    @pytest.mark.anyio
+    async def test_eager_load_survives_commit(self, mixin_session):
+        """expire_on_commit=False: the reload must not expire the relation."""
+        owner = await self._seed_eager(mixin_session)
+
+        await mixin_session.commit()
+
+        assert "target" not in sa_inspect(owner).unloaded
+        assert owner.target.name == "t"
+
+    @pytest.mark.anyio
+    async def test_eager_load_survives_commit_expire_on_commit(
+        self, mixin_session_expire
+    ):
+        """expire_on_commit=True: what was loaded must be recorded before the commit."""
+        owner = await self._seed_eager(mixin_session_expire)
+
+        await mixin_session_expire.commit()
+
+        assert "target" not in sa_inspect(owner).unloaded
+        assert owner.target.name == "t"
+
+    @pytest.mark.anyio
+    async def test_unloaded_relation_stays_unloaded(self, mixin_session):
+        """Only what was loaded is restored: the reload must not eager-load extra."""
+        target = RelTarget(name="t")
+        mixin_session.add(target)
+        await mixin_session.flush()
+        owner = RelOwner(title="o", target_id=target.id)
+        mixin_session.add(owner)
+
+        await mixin_session.commit()
+
+        assert "target" in sa_inspect(owner).unloaded
+
+
 class TestUUIDMixin:
     @pytest.mark.anyio
     async def test_uuid_generated_by_db(self, mixin_session):
@@ -1013,10 +1097,10 @@ class TestEventCallbacks:
 
         real_batch_reload = _watched_module._batch_reload
 
-        async def racing_batch_reload(session, model, pk_tuples):
-            if any(pk[0] == doomed_id for pk in pk_tuples):
+        async def racing_batch_reload(session, model, objs, preloaded):
+            if any(getattr(o, "id", None) == doomed_id for o in objs):
                 await kill_doomed_row_once()
-            return await real_batch_reload(session, model, pk_tuples)
+            return await real_batch_reload(session, model, objs, preloaded)
 
         # Patch the batched reload EventSession.commit() uses to pick up
         # server defaults, so this test still exercises the race.
@@ -1039,7 +1123,7 @@ class TestEventCallbacks:
         obj = WatchedModel(status="active", other="x")
         mixin_session.add(obj)
 
-        async def failing_batch_reload(session, model, pk_tuples):
+        async def failing_batch_reload(session, model, objs, preloaded):
             raise RuntimeError("reload failed")
 
         with (

uv.lock

@@ -973,15 +973,15 @@ wheels = [
 
 [[package]]
 name = "pymdown-extensions"
-version = "11.0"
+version = "11.0.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "markdown" },
     { name = "pyyaml" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/47/67/f1e79672a5f91985577c7984c9709ca110e4fd37fe7fd167b60422e6ccc2/pymdown_extensions-11.0.tar.gz", hash = "sha256:8269cef0247f9e2d0a62fcea10860aba05c1cbab5470fd4b63230b96434dc589", size = 857049, upload-time = "2026-06-23T02:27:45.146Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/21/a9/5f0c535ba3b08fe09270c16808e053a968868242ecbd5676d4e3a488bf28/pymdown_extensions-11.0.1.tar.gz", hash = "sha256:dd2905ae6fc5b75582fafb139a1266ffc754705efa902aa50067fa7ff4f94ec0", size = 857113, upload-time = "2026-07-02T17:59:22.955Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/af/b6/1ae53367e28b9cffa3be7574e13fbe4589694272fd47710fbdbafd3d63c6/pymdown_extensions-11.0-py3-none-any.whl", hash = "sha256:fbc4acb641814fa9d17521bbd21a5240ef739a662f11c06330c4b78c93e954d6", size = 269415, upload-time = "2026-06-23T02:27:43.826Z" },
+    { url = "https://files.pythonhosted.org/packages/d6/54/da572c98c0b77626a91b5d3b89f0231d8bff5125c225420908632f8b342d/pymdown_extensions-11.0.1-py3-none-any.whl", hash = "sha256:db3943a62bab7e03af1364f0c4083e64b91fb097675a4b6cceccfbe9a77e5eb2", size = 269455, upload-time = "2026-07-02T17:59:21.271Z" },
 ]
 
 [[package]]