feat: add include_total flag to offset pagination to skip COUNT query (#158)

docs/examples/pagination-search.md

@@ -85,6 +85,8 @@ GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published&or
 
 `filter_attributes` always reflects the values visible **after** applying the active filters. Use it to populate filter dropdowns on the client.
 
+To skip the `COUNT(*)` query for better performance on large tables, pass `include_total=False`. `pagination.total_count` will be `null` in the response, while `has_more` remains accurate.
+
 ### Cursor pagination
 
 Best for feeds, infinite scroll, or any high-throughput API where offset performance degrades.

docs/module/crud.md

@@ -189,6 +189,22 @@ The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.Async
 }
 ```
 
+#### Skipping the COUNT query
+
+!!! info "Added in `v2.4.1`"
+
+By default `offset_paginate` runs two queries: one for the page items and one `COUNT(*)` for `total_count`. On large tables the `COUNT` can be expensive. Pass `include_total=False` to skip it:
+
+```python
+result = await UserCrud.offset_paginate(
+    session=session,
+    page=page,
+    items_per_page=items_per_page,
+    include_total=False,
+    schema=UserRead,
+)
+```
+
 ### Cursor pagination
 
 ```python

src/fastapi_toolsets/crud/factory.py

@@ -922,6 +922,7 @@ class AsyncCrud(Generic[ModelType]):
         order_by: OrderByClause | None = None,
         page: int = 1,
         items_per_page: int = 20,
+        include_total: bool = True,
         search: str | SearchConfig | None = None,
         search_fields: Sequence[SearchFieldType] | None = None,
         facet_fields: Sequence[FacetFieldType] | None = None,
@@ -939,6 +940,8 @@ class AsyncCrud(Generic[ModelType]):
             order_by: Column or list of columns to order by
             page: Page number (1-indexed)
             items_per_page: Number of items per page
+            include_total: When ``False``, skip the ``COUNT`` query;
+                ``pagination.total_count`` will be ``None``.
             search: Search query string or SearchConfig object
             search_fields: Fields to search in (overrides class default)
             facet_fields: Columns to compute distinct values for (overrides class default)
@@ -983,28 +986,39 @@ class AsyncCrud(Generic[ModelType]):
         if order_by is not None:
             q = q.order_by(order_by)
 
-        q = q.offset(offset).limit(items_per_page)
-        result = await session.execute(q)
-        raw_items = cast(list[ModelType], result.unique().scalars().all())
+        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 = count_q.select_from(cls.model)
+
+            # Apply explicit joins to count query
+            count_q = _apply_joins(count_q, joins, outer_join)
+
+            # Apply search joins to count query
+            count_q = _apply_search_joins(count_q, search_joins)
+
+            if filters:
+                count_q = count_q.where(and_(*filters))
+
+            count_result = await session.execute(count_q)
+            total_count: int | None = 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
+
         items: list[Any] = [schema.model_validate(item) for item in raw_items]
 
-        # 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 = count_q.select_from(cls.model)
-
-        # Apply explicit joins to count query
-        count_q = _apply_joins(count_q, joins, outer_join)
-
-        # Apply search joins to count query
-        count_q = _apply_search_joins(count_q, search_joins)
-
-        if filters:
-            count_q = count_q.where(and_(*filters))
-
-        count_result = await session.execute(count_q)
-        total_count = count_result.scalar_one()
-
         filter_attributes = await cls._build_filter_attributes(
             session, facet_fields, filters, search_joins
         )
@@ -1015,7 +1029,7 @@ class AsyncCrud(Generic[ModelType]):
                 total_count=total_count,
                 items_per_page=items_per_page,
                 page=page,
-                has_more=page * items_per_page < total_count,
+                has_more=has_more,
             ),
             filter_attributes=filter_attributes,
         )
@@ -1190,6 +1204,7 @@ class AsyncCrud(Generic[ModelType]):
         page: int = ...,
         cursor: str | None = ...,
         items_per_page: int = ...,
+        include_total: bool = ...,
         search: str | SearchConfig | None = ...,
         search_fields: Sequence[SearchFieldType] | None = ...,
         facet_fields: Sequence[FacetFieldType] | None = ...,
@@ -1212,6 +1227,7 @@ class AsyncCrud(Generic[ModelType]):
         page: int = ...,
         cursor: str | None = ...,
         items_per_page: int = ...,
+        include_total: bool = ...,
         search: str | SearchConfig | None = ...,
         search_fields: Sequence[SearchFieldType] | None = ...,
         facet_fields: Sequence[FacetFieldType] | None = ...,
@@ -1233,6 +1249,7 @@ class AsyncCrud(Generic[ModelType]):
         page: int = 1,
         cursor: str | None = None,
         items_per_page: int = 20,
+        include_total: bool = True,
         search: str | SearchConfig | None = None,
         search_fields: Sequence[SearchFieldType] | None = None,
         facet_fields: Sequence[FacetFieldType] | None = None,
@@ -1258,6 +1275,8 @@ class AsyncCrud(Generic[ModelType]):
                 :class:`.CursorPaginatedResponse`.  Only used when
                 ``pagination_type`` is ``CURSOR``.
             items_per_page: Number of items per page (default 20).
+            include_total: When ``False``, skip the ``COUNT`` query;
+                only applies when ``pagination_type`` is ``OFFSET``.
             search: Search query string or :class:`.SearchConfig` object.
             search_fields: Fields to search in (overrides class default).
             facet_fields: Columns to compute distinct values for (overrides
@@ -1304,6 +1323,7 @@ class AsyncCrud(Generic[ModelType]):
                     order_by=order_by,
                     page=page,
                     items_per_page=items_per_page,
+                    include_total=include_total,
                     search=search,
                     search_fields=search_fields,
                     facet_fields=facet_fields,

src/fastapi_toolsets/schemas.py

@@ -98,13 +98,14 @@ class OffsetPagination(PydanticBase):
     """Pagination metadata for offset-based list responses.
 
     Attributes:
-        total_count: Total number of items across all pages
+        total_count: Total number of items across all pages.
+            ``None`` when ``include_total=False``.
         items_per_page: Number of items per page
         page: Current page number (1-indexed)
         has_more: Whether there are more pages
     """
 
-    total_count: int
+    total_count: int | None
     items_per_page: int
     page: int
     has_more: bool

tests/test_crud.py

@@ -1759,6 +1759,52 @@ class TestSchemaResponse:
         assert result.data[0].username == "pg_user"
         assert not hasattr(result.data[0], "email")
 
+    @pytest.mark.anyio
+    async def test_include_total_false_skips_count(self, db_session: AsyncSession):
+        """offset_paginate with include_total=False returns total_count=None."""
+        from fastapi_toolsets.schemas import OffsetPagination
+
+        for i in range(5):
+            await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
+
+        result = await RoleCrud.offset_paginate(
+            db_session, items_per_page=10, include_total=False, schema=RoleRead
+        )
+
+        assert isinstance(result.pagination, OffsetPagination)
+        assert result.pagination.total_count is None
+        assert len(result.data) == 5
+        assert result.pagination.has_more is False
+
+    @pytest.mark.anyio
+    async def test_include_total_false_has_more_true(self, db_session: AsyncSession):
+        """offset_paginate with include_total=False sets has_more via extra-row probe."""
+        for i in range(15):
+            await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
+
+        result = await RoleCrud.offset_paginate(
+            db_session, items_per_page=10, include_total=False, schema=RoleRead
+        )
+
+        assert result.pagination.total_count is None
+        assert result.pagination.has_more is True
+        assert len(result.data) == 10
+
+    @pytest.mark.anyio
+    async def test_include_total_false_exact_page_boundary(
+        self, db_session: AsyncSession
+    ):
+        """offset_paginate with include_total=False: has_more=False when items == page size."""
+        for i in range(10):
+            await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
+
+        result = await RoleCrud.offset_paginate(
+            db_session, items_per_page=10, include_total=False, schema=RoleRead
+        )
+
+        assert result.pagination.has_more is False
+        assert len(result.data) == 10
+
 
 class TestCursorPaginate:
     """Tests for cursor-based pagination via cursor_paginate()."""
@@ -2521,3 +2567,20 @@ class TestPaginate:
                 pagination_type="unknown",
                 schema=RoleRead,
             )  # type: ignore[no-matching-overload]
+
+    @pytest.mark.anyio
+    async def test_offset_include_total_false(self, db_session: AsyncSession):
+        """paginate() passes include_total=False through to offset_paginate."""
+        from fastapi_toolsets.schemas import OffsetPagination
+
+        await RoleCrud.create(db_session, RoleCreate(name="admin"))
+
+        result = await RoleCrud.paginate(
+            db_session,
+            pagination_type=PaginationType.OFFSET,
+            include_total=False,
+            schema=RoleRead,
+        )
+
+        assert isinstance(result.pagination, OffsetPagination)
+        assert result.pagination.total_count is None

tests/test_schemas.py

@@ -201,6 +201,27 @@ class TestOffsetPagination:
         assert data["page"] == 2
         assert data["has_more"] is True
 
+    def test_total_count_can_be_none(self):
+        """total_count accepts None (include_total=False mode)."""
+        pagination = OffsetPagination(
+            total_count=None,
+            items_per_page=20,
+            page=1,
+            has_more=True,
+        )
+        assert pagination.total_count is None
+
+    def test_serialization_with_none_total_count(self):
+        """OffsetPagination serializes total_count=None correctly."""
+        pagination = OffsetPagination(
+            total_count=None,
+            items_per_page=20,
+            page=1,
+            has_more=False,
+        )
+        data = pagination.model_dump()
+        assert data["total_count"] is None
+
 
 class TestCursorPagination:
     """Tests for CursorPagination schema."""