codekingpro/portable-devtools

1from __future__ import annotations2 3import typing as t4from math import ceil5 6import sqlalchemy as sa7import sqlalchemy.orm as sa_orm8from flask import abort9from flask import request10 11 12class Pagination:13    """Apply an offset and limit to the query based on the current page and number of14    items per page.15 16    Don't create pagination objects manually. They are created by17    :meth:`.SQLAlchemy.paginate` and :meth:`.Query.paginate`.18 19    This is a base class, a subclass must implement :meth:`_query_items` and20    :meth:`_query_count`. Those methods will use arguments passed as ``kwargs`` to21    perform the queries.22 23    :param page: The current page, used to calculate the offset. Defaults to the24        ``page`` query arg during a request, or 1 otherwise.25    :param per_page: The maximum number of items on a page, used to calculate the26        offset and limit. Defaults to the ``per_page`` query arg during a request,27        or 20 otherwise.28    :param max_per_page: The maximum allowed value for ``per_page``, to limit a29        user-provided value. Use ``None`` for no limit. Defaults to 100.30    :param error_out: Abort with a ``404 Not Found`` error if no items are returned31        and ``page`` is not 1, or if ``page`` or ``per_page`` is less than 1, or if32        either are not ints.33    :param count: Calculate the total number of values by issuing an extra count34        query. For very complex queries this may be inaccurate or slow, so it can be35        disabled and set manually if necessary.36    :param kwargs: Information about the query to paginate. Different subclasses will37        require different arguments.38 39    .. versionchanged:: 3.040        Iterating over a pagination object iterates over its items.41 42    .. versionchanged:: 3.043        Creating instances manually is not a public API.44    """45 46    def __init__(47        self,48        page: int | None = None,49        per_page: int | None = None,50        max_per_page: int | None = 100,51        error_out: bool = True,52        count: bool = True,53        **kwargs: t.Any,54    ) -> None:55        self._query_args = kwargs56        page, per_page = self._prepare_page_args(57            page=page,58            per_page=per_page,59            max_per_page=max_per_page,60            error_out=error_out,61        )62 63        self.page: int = page64        """The current page."""65 66        self.per_page: int = per_page67        """The maximum number of items on a page."""68 69        self.max_per_page: int | None = max_per_page70        """The maximum allowed value for ``per_page``."""71 72        items = self._query_items()73 74        if not items and page != 1 and error_out:75            abort(404)76 77        self.items: list[t.Any] = items78        """The items on the current page. Iterating over the pagination object is79        equivalent to iterating over the items.80        """81 82        if count:83            total = self._query_count()84        else:85            total = None86 87        self.total: int | None = total88        """The total number of items across all pages."""89 90    @staticmethod91    def _prepare_page_args(92        *,93        page: int | None = None,94        per_page: int | None = None,95        max_per_page: int | None = None,96        error_out: bool = True,97    ) -> tuple[int, int]:98        if request:99            if page is None:100                try:101                    page = int(request.args.get("page", 1))102                except (TypeError, ValueError):103                    if error_out:104                        abort(404)105 106                    page = 1107 108            if per_page is None:109                try:110                    per_page = int(request.args.get("per_page", 20))111                except (TypeError, ValueError):112                    if error_out:113                        abort(404)114 115                    per_page = 20116        else:117            if page is None:118                page = 1119 120            if per_page is None:121                per_page = 20122 123        if max_per_page is not None:124            per_page = min(per_page, max_per_page)125 126        if page < 1:127            if error_out:128                abort(404)129            else:130                page = 1131 132        if per_page < 1:133            if error_out:134                abort(404)135            else:136                per_page = 20137 138        return page, per_page139 140    @property141    def _query_offset(self) -> int:142        """The index of the first item to query, passed to ``offset()``.143 144        :meta private:145 146        .. versionadded:: 3.0147        """148        return (self.page - 1) * self.per_page149 150    def _query_items(self) -> list[t.Any]:151        """Execute the query to get the items on the current page.152 153        Uses init arguments stored in :attr:`_query_args`.154 155        :meta private:156 157        .. versionadded:: 3.0158        """159        raise NotImplementedError160 161    def _query_count(self) -> int:162        """Execute the query to get the total number of items.163 164        Uses init arguments stored in :attr:`_query_args`.165 166        :meta private:167 168        .. versionadded:: 3.0169        """170        raise NotImplementedError171 172    @property173    def first(self) -> int:174        """The number of the first item on the page, starting from 1, or 0 if there are175        no items.176 177        .. versionadded:: 3.0178        """179        if len(self.items) == 0:180            return 0181 182        return (self.page - 1) * self.per_page + 1183 184    @property185    def last(self) -> int:186        """The number of the last item on the page, starting from 1, inclusive, or 0 if187        there are no items.188 189        .. versionadded:: 3.0190        """191        first = self.first192        return max(first, first + len(self.items) - 1)193 194    @property195    def pages(self) -> int:196        """The total number of pages."""197        if self.total == 0 or self.total is None:198            return 0199 200        return ceil(self.total / self.per_page)201 202    @property203    def has_prev(self) -> bool:204        """``True`` if this is not the first page."""205        return self.page > 1206 207    @property208    def prev_num(self) -> int | None:209        """The previous page number, or ``None`` if this is the first page."""210        if not self.has_prev:211            return None212 213        return self.page - 1214 215    def prev(self, *, error_out: bool = False) -> Pagination:216        """Query the :class:`Pagination` object for the previous page.217 218        :param error_out: Abort with a ``404 Not Found`` error if no items are returned219            and ``page`` is not 1, or if ``page`` or ``per_page`` is less than 1, or if220            either are not ints.221        """222        p = type(self)(223            page=self.page - 1,224            per_page=self.per_page,225            error_out=error_out,226            count=False,227            **self._query_args,228        )229        p.total = self.total230        return p231 232    @property233    def has_next(self) -> bool:234        """``True`` if this is not the last page."""235        return self.page < self.pages236 237    @property238    def next_num(self) -> int | None:239        """The next page number, or ``None`` if this is the last page."""240        if not self.has_next:241            return None242 243        return self.page + 1244 245    def next(self, *, error_out: bool = False) -> Pagination:246        """Query the :class:`Pagination` object for the next page.247 248        :param error_out: Abort with a ``404 Not Found`` error if no items are returned249            and ``page`` is not 1, or if ``page`` or ``per_page`` is less than 1, or if250            either are not ints.251        """252        p = type(self)(253            page=self.page + 1,254            per_page=self.per_page,255            max_per_page=self.max_per_page,256            error_out=error_out,257            count=False,258            **self._query_args,259        )260        p.total = self.total261        return p262 263    def iter_pages(264        self,265        *,266        left_edge: int = 2,267        left_current: int = 2,268        right_current: int = 4,269        right_edge: int = 2,270    ) -> t.Iterator[int | None]:271        """Yield page numbers for a pagination widget. Skipped pages between the edges272        and middle are represented by a ``None``.273 274        For example, if there are 20 pages and the current page is 7, the following275        values are yielded.276 277        .. code-block:: python278 279            1, 2, None, 5, 6, 7, 8, 9, 10, 11, None, 19, 20280 281        :param left_edge: How many pages to show from the first page.282        :param left_current: How many pages to show left of the current page.283        :param right_current: How many pages to show right of the current page.284        :param right_edge: How many pages to show from the last page.285 286        .. versionchanged:: 3.0287            Improved efficiency of calculating what to yield.288 289        .. versionchanged:: 3.0290            ``right_current`` boundary is inclusive.291 292        .. versionchanged:: 3.0293            All parameters are keyword-only.294        """295        pages_end = self.pages + 1296 297        if pages_end == 1:298            return299 300        left_end = min(1 + left_edge, pages_end)301        yield from range(1, left_end)302 303        if left_end == pages_end:304            return305 306        mid_start = max(left_end, self.page - left_current)307        mid_end = min(self.page + right_current + 1, pages_end)308 309        if mid_start - left_end > 0:310            yield None311 312        yield from range(mid_start, mid_end)313 314        if mid_end == pages_end:315            return316 317        right_start = max(mid_end, pages_end - right_edge)318 319        if right_start - mid_end > 0:320            yield None321 322        yield from range(right_start, pages_end)323 324    def __iter__(self) -> t.Iterator[t.Any]:325        yield from self.items326 327 328class SelectPagination(Pagination):329    """Returned by :meth:`.SQLAlchemy.paginate`. Takes ``select`` and ``session``330    arguments in addition to the :class:`Pagination` arguments.331 332    .. versionadded:: 3.0333    """334 335    def _query_items(self) -> list[t.Any]:336        select = self._query_args["select"]337        select = select.limit(self.per_page).offset(self._query_offset)338        session = self._query_args["session"]339        return list(session.execute(select).unique().scalars())340 341    def _query_count(self) -> int:342        select = self._query_args["select"]343        sub = select.options(sa_orm.lazyload("*")).order_by(None).subquery()344        session = self._query_args["session"]345        out = session.execute(sa.select(sa.func.count()).select_from(sub)).scalar()346        return out  # type: ignore[no-any-return]347 348 349class QueryPagination(Pagination):350    """Returned by :meth:`.Query.paginate`. Takes a ``query`` argument in addition to351    the :class:`Pagination` arguments.352 353    .. versionadded:: 3.0354    """355 356    def _query_items(self) -> list[t.Any]:357        query = self._query_args["query"]358        out = query.limit(self.per_page).offset(self._query_offset).all()359        return out  # type: ignore[no-any-return]360 361    def _query_count(self) -> int:362        # Query.count automatically disables eager loads363        out = self._query_args["query"].order_by(None).count()364        return out  # type: ignore[no-any-return]365