Add paginate_rows() method

Author: jwodder

CHANGES.rst

@@ -1,3 +1,11 @@
+Version 3.2.0
+-------------
+
+Unreleased
+
+-   Added ``paginate_rows`` method to the extension object for paginating over
+    ``Row`` objects :issue:`1168`:
+
 Version 3.1.1
 -------------
 

docs/api.rst

@@ -65,7 +65,8 @@ Pagination
     based on the current page and number of items per page.
 
     Don't create pagination objects manually. They are created by
-    :meth:`.SQLAlchemy.paginate` and :meth:`.Query.paginate`.
+    :meth:`.SQLAlchemy.paginate`, :meth:`.SQLAlchemy.paginate_rows`, and
+    :meth:`.Query.paginate`.
 
     .. versionchanged:: 3.0
         Iterating over a pagination object iterates over its items.

src/flask_sqlalchemy/extension.py

@@ -22,6 +22,7 @@
 from .model import Model
 from .model import NameMixin
 from .pagination import Pagination
+from .pagination import RowPagination
 from .pagination import SelectPagination
 from .query import Query
 from .session import _app_ctx_id
@@ -814,7 +815,8 @@ def paginate(
 
         The statement should select a model class, like ``select(User)``. This applies
         ``unique()`` and ``scalars()`` modifiers to the result, so compound selects will
-        not return the expected results.
+        not return the expected results.  To paginate a compound select, use
+        :meth:`paginate_rows` instead.
 
         :param select: The ``select`` statement to paginate.
         :param page: The current page, used to calculate the offset. Defaults to the
@@ -846,6 +848,54 @@ def paginate(
             count=count,
         )
 
+    def paginate_rows(
+        self,
+        select: sa.sql.Select[t.Any],
+        *,
+        page: int | None = None,
+        per_page: int | None = None,
+        max_per_page: int | None = None,
+        error_out: bool = True,
+        count: bool = True,
+    ) -> Pagination:
+        """Apply an offset and limit to a select statment based on the current page and
+        number of items per page, returning a :class:`.Pagination` object.
+
+        Unlike :meth:`paginate`, the statement may select any number of
+        columns, like ``select(User.name, User.password)``.  Regardless of how
+        many columns are selected, the :attr:`.Pagination.items` attribute of
+        the returned :class:`.Pagination` instance will contain :class:`Row
+        <sqlalchemy.engine.Row>` objects.
+
+        Note that the ``unique()`` modifier is applied to the result.
+
+        :param select: The ``select`` statement to paginate.
+        :param page: The current page, used to calculate the offset. Defaults to the
+            ``page`` query arg during a request, or 1 otherwise.
+        :param per_page: The maximum number of items on a page, used to calculate the
+            offset and limit. Defaults to the ``per_page`` query arg during a request,
+            or 20 otherwise.
+        :param max_per_page: The maximum allowed value for ``per_page``, to limit a
+            user-provided value. Use ``None`` for no limit. Defaults to 100.
+        :param error_out: Abort with a ``404 Not Found`` error if no items are returned
+            and ``page`` is not 1, or if ``page`` or ``per_page`` is less than 1, or if
+            either are not ints.
+        :param count: Calculate the total number of values by issuing an extra count
+            query. For very complex queries this may be inaccurate or slow, so it can be
+            disabled and set manually if necessary.
+
+        .. versionadded:: 3.2
+        """
+        return RowPagination(
+            select=select,
+            session=self.session(),
+            page=page,
+            per_page=per_page,
+            max_per_page=max_per_page,
+            error_out=error_out,
+            count=count,
+        )
+
     def _call_for_binds(
         self, bind_key: str | None | list[str | None], op_name: str
     ) -> None:

src/flask_sqlalchemy/pagination.py

@@ -14,7 +14,8 @@ class Pagination:
     items per page.
 
     Don't create pagination objects manually. They are created by
-    :meth:`.SQLAlchemy.paginate` and :meth:`.Query.paginate`.
+    :meth:`.SQLAlchemy.paginate`, :meth:`.SQLAlchemy.paginate_rows`, and
+    :meth:`.Query.paginate`.
 
     This is a base class, a subclass must implement :meth:`_query_items` and
     :meth:`_query_count`. Those methods will use arguments passed as ``kwargs`` to
@@ -346,6 +347,28 @@ def _query_count(self) -> int:
         return out  # type: ignore[no-any-return]
 
 
+class RowPagination(Pagination):
+    """Returned by :meth:`.SQLAlchemy.paginate_rows`. Takes ``select`` and ``session``
+    arguments in addition to the :class:`Pagination` arguments.
+
+    .. versionadded:: 3.2
+    """
+
+    def _query_items(self) -> list[t.Any]:
+        # Like SelectPagination._query_items(), but without the `.scalars()`
+        select = self._query_args["select"]
+        select = select.limit(self.per_page).offset(self._query_offset)
+        session = self._query_args["session"]
+        return list(session.execute(select).unique())
+
+    def _query_count(self) -> int:
+        select = self._query_args["select"]
+        sub = select.options(sa_orm.lazyload("*")).order_by(None).subquery()
+        session = self._query_args["session"]
+        out = session.execute(sa.select(sa.func.count()).select_from(sub)).scalar()
+        return out  # type: ignore[no-any-return]
+
+
 class QueryPagination(Pagination):
     """Returned by :meth:`.Query.paginate`. Takes a ``query`` argument in addition to
     the :class:`Pagination` arguments.

tests/test_pagination.py

@@ -3,6 +3,7 @@
 import typing as t
 
 import pytest
+import sqlalchemy as sa
 from flask import Flask
 from werkzeug.exceptions import NotFound
 
@@ -158,6 +159,8 @@ def test_paginate(paginate: _PaginateCallable) -> None:
     assert p.page == 1
     assert p.per_page == 20
     assert len(p.items) == 20
+    for it in p.items:
+        assert isinstance(it, paginate.Todo)
     assert p.total == 250
     assert p.pages == 13
 
@@ -203,3 +206,52 @@ def test_no_items_404(db: SQLAlchemy, Todo: t.Any) -> None:
 
     with pytest.raises(NotFound):
         db.paginate(db.select(Todo), page=2)
+
+
+class _RowPaginateCallable:
+    def __init__(self, app: Flask, db: SQLAlchemy, Todo: t.Any) -> None:
+        self.app = app
+        self.db = db
+        self.Todo = Todo
+
+    def __call__(
+        self,
+        page: int | None = None,
+        per_page: int | None = None,
+        max_per_page: int | None = None,
+        error_out: bool = True,
+        count: bool = True,
+    ) -> Pagination:
+        qs = {"page": page, "per_page": per_page}
+        with self.app.test_request_context(query_string=qs):
+            return self.db.paginate_rows(
+                self.db.select(self.Todo.id, self.Todo.title),
+                max_per_page=max_per_page,
+                error_out=error_out,
+                count=count,
+            )
+
+
+@pytest.fixture
+def paginate_rows(app: Flask, db: SQLAlchemy, Todo: t.Any) -> _RowPaginateCallable:
+    with app.app_context():
+        for i in range(1, 251):
+            db.session.add(Todo(title=f"task {i}"))
+
+        db.session.commit()
+
+    return _RowPaginateCallable(app, db, Todo)
+
+
+def test_paginate_rows(paginate_rows: _RowPaginateCallable) -> None:
+    p = paginate_rows()
+    assert p.page == 1
+    assert p.per_page == 20
+    assert len(p.items) == 20
+    for it in p.items:
+        assert isinstance(it, sa.Row)
+        assert len(it) == 2
+        assert isinstance(it[0], int)
+        assert isinstance(it[1], str)
+    assert p.total == 250
+    assert p.pages == 13