From 50529b1081485d37f2694baf2955322d4b99c992 Mon Sep 17 00:00:00 2001 From: d3vyce Date: Fri, 28 Aug 2026 18:01:31 -0400 Subject: [PATCH] fix: pagination over a to-many join duplicates, drops and truncates results --- src/fastapi_toolsets/crud/factory.py | 186 ++++++++++++++++++---- tests/test_crud_search.py | 226 ++++++++++++++++++++++++++- 2 files changed, 383 insertions(+), 29 deletions(-) diff --git a/src/fastapi_toolsets/crud/factory.py b/src/fastapi_toolsets/crud/factory.py index bd572c0..baaa3ee 100644 --- a/src/fastapi_toolsets/crud/factory.py +++ b/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) - - # Fetch one extra to detect whether another page exists in this direction - q = q.limit(items_per_page + 1) - result = await session.execute(q) - raw_items = cast(list[ModelType], result.unique().scalars().all()) + order_clauses.append(order_by) + q = q.order_by(*order_clauses) + # 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] diff --git a/tests/test_crud_search.py b/tests/test_crud_search.py index a8594f6..beb3ae1 100644 --- a/tests/test_crud_search.py +++ b/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, ) @@ -397,6 +402,225 @@ class TestBuildSearchFilters: 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."""