Question: how to change the parameters used for pagination?

[cydergoth]

Our UI team wants to use these parameters for pagination: https://github.com/typicode/json-server#paginate

I can see in the docs that it says you can override the methods to change these parameters, but I don't see an example and it looks like it is in a mixin, which I'm not sure how to override.

Any advice gratefully received!

[lafrech]

PaginationMixin is inherited by Blueprint so just override what you want in a custom Blueprint.

Looking at it, I realize that the pagination feature is not that easy to customize. Some parameters are exposed to be easy to modify, but other than that, I don't see an easy way to modify PaginationParametersSchema without duplicating the paginate decorator.

I'm open to suggestions about how to make things easier.

We could remove _pagination_parameters_schema_factory and set PaginationParametersSchema as class attribute instead of DEFAULT_PAGINATION_PARAMETERS. It would be slightly more verbose to override pagination parameters but it would make it easy to do other modifications, like renaming parameters, etc. But then we wouldn't be able to override those parameters in the call to paginate.

Another option would be to allow the user to customize the factory to create the schema they want while still allowing to pass modifiers like page size, etc.

[cydergoth]

Would it be possible to implement a set of mixins for the protocol in the above link? That would make Smorest work smoothly with next.js and React-Admin

[cydergoth]

Maybe have the whole pagination mixin be a shim to a user-provided impl?

[cydergoth]

I implemented this:

"""Slice feature

Two Slice modes are supported:

- Slice inside the resource: the resource function is responsible for
  selecting requested range of items and setting total number of items.

- Post-Slice: the resource returns an iterator (typically a DB cursor) and
  a pager is provided to paginate the data and get the total number of items.
"""
from copy import deepcopy
from functools import wraps
import http
import json
import warnings
import sys

from flask import request

import marshmallow as ma
from webargs.flaskparser import FlaskParser

from flask_smorest.utils import unpack_tuple_response


class SliceParameters:
    """Holds Slice arguments

    :param int start: Slice start (inclusive)
    :param int end: Slice end (exclusive)
    """

    def __init__(self, start, end ):
        self.start = start
        self.end = end
        self.item_count = None

    @property
    def first_item(self):
        """Return first item number"""
        return self.start

    @property
    def last_item(self):
        """Return last item number"""
        return self.end

    def __repr__(self):
        return "{}(_start={!r},_end={!r})".format(
            self.__class__.__name__, self.start, self.end
        )


def _slice_parameters_schema_factory( def_start, def_end):
    """Generate a SliceParametersSchema"""

    class SliceParametersSchema(ma.Schema):
        """Deserializes slice params into SliceParameters"""

        class Meta:
            ordered = True
            unknown = ma.EXCLUDE

        start = ma.fields.Integer(data_key="_start",
            load_default=def_start, validate=ma.validate.Range(min=1)
        )
        end = ma.fields.Integer(data_key="_end",
            load_default=def_end)
            #validate=ma.validate.Range(min=1, max=def_max_page_size),


        @ma.post_load
        def make_slicer(self, data, **kwargs):
            return SliceParameters(**data)

    return SliceParametersSchema


class Slice:
    """Pager for simple types such as lists.

    Can be subclassed to provide a pager for a specific data object.
    """

    def __init__(self, collection, slice_params):
        """Create a Slice instance

        :param sequence collection: Collection of items to page through
        :page SliceParameters slice_params: Slice parameters
        """
        self.collection = collection
        self.slice_params = slice_params
        self.slice_params.item_count = self.item_count

    @property
    def items(self):
        return list(
            self.collection[
                self.slice_params.first_item : self.slice_params.last_item + 1
            ]
        )

    @property
    def item_count(self):
        return len(self.collection)

    def __repr__(self):
        return "{}(collection={!r},page_params={!r})".format(
            self.__class__.__name__, self.collection, self.slice_params
        )


class SliceMetadataSchema(ma.Schema):
    """Slice metadata schema

    Used to serialize Slice metadata.
    Its main purpose is to document the Slice metadata.
    """

    total = ma.fields.Int()
    start = ma.fields.Int()
    end = ma.fields.Int()

    class Meta:
        ordered = True


SLICE_HEADER = {
    "description": "Slice metadata",
    "schema": SliceMetadataSchema,
}


class SliceMixin:
    """Extend Blueprint to add Slice feature"""

    SLICE_ARGUMENTS_PARSER = FlaskParser()

    # Name of field to use for Slice metadata response header
    # Can be overridden. If None, no Slice header is returned.
    SLICE_HEADER_NAME = "X-Slice"

    # Global default Slice parameters
    # Can be overridden to provide custom defaults
    DEFAULT_SLICE_PARAMETERS = {"start": 1, "end": 10 }

    def slice(self, pager=None, *, start=None, end=None):
        """Decorator adding Slice to the endpoint

        :param Page pager: Page class used to paginate response data
        :param int start: Default requested start index (default: 1)
        :param int end: Default requested end index (default: 10)

        If a :class:`Page <Page>` class is provided, it is used to paginate the
        data returned by the view function, typically a lazy database cursor.

        Otherwise, Slice is handled in the view function.

        The decorated function may return a tuple including status and/or
        headers, like a typical flask view function. It may not return a
        ``Response`` object.

        See :doc:`Slice <Slice>`.
        """
        if start is None:
            start = self.DEFAULT_SLICE_PARAMETERS["start"]
        if end is None:
            end = self.DEFAULT_SLICE_PARAMETERS["end"]
        slice_params_schema = _slice_parameters_schema_factory(
            start, end
        )

        parameters = {
            "in": "query",
            "schema": slice_params_schema,
        }

        error_status_code = self.SLICE_ARGUMENTS_PARSER.DEFAULT_VALIDATION_STATUS

        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                slice_params = self.SLICE_ARGUMENTS_PARSER.parse(
                    slice_params_schema, request, location="query"
                )

                # Slice in resource code: inject page_params as kwargs
                if pager is None:
                    kwargs["slice_parameters"] = slice_params

                # Execute decorated function
                result, status, headers = unpack_tuple_response(func(*args, **kwargs))

                # Post Slice: use pager class to paginate the result
                if pager is not None:
                    result = pager(result, slice_params=slice_params).items

                # Set Slice metadata in response
                if self.SLICE_HEADER_NAME is not None:
                    if slice_params.item_count is None:
                        warnings.warn(
                            "item_count not set in endpoint {}.".format(
                                request.endpoint
                            )
                        )
                    else:
                        result, headers = self._set_slice_metadata(
                            slice_params, result, headers
                        )

                return result, status, headers

            # Add Slice params to doc info in wrapper object
            wrapper._apidoc = deepcopy(getattr(wrapper, "_apidoc", {}))
            wrapper._apidoc["slice"] = {
                "parameters": parameters,
                "response": {
                    error_status_code: http.HTTPStatus(error_status_code).name,
                },
            }

            return wrapper

        return decorator

    @staticmethod
    def _make_slice_metadata(start, end, item_count):
        """Build Slice metadata from page, page size and item count

        Override this to use another Slice metadata structure
        """
        slice_metadata = {}
        slice_metadata["total"] = item_count
        if item_count == 0:
            slice_metadata["total"] = 0
        else:
            # First / last slice, slice count
            slice_metadata["start"] = start
            slice_metadata["end"] = end
        return SliceMetadataSchema().dump(slice_metadata)

    def _set_slice_metadata(self, slice_params, result, headers):
        """Add Slice metadata to headers

        Override this to set Slice data another way
        """
        if headers is None:
            headers = {}
        headers[self.SLICE_HEADER_NAME] = json.dumps(
            self._make_slice_metadata(
                slice_params.start, slice_params.end, slice_params.item_count
            )
        )
        return result, headers

    def _document_slice_metadata(self, spec, resp_doc):
        """Document Slice metadata header

        Override this to document custom Slice metadata
        """
        resp_doc["headers"] = {
            self.SLICE_HEADER_NAME: "SLICE"
            if spec.openapi_version.major >= 3
            else SLICE_HEADER
        }

    def _prepare_slice_doc(self, doc, doc_info, *, spec, **kwargs):
        operation = doc_info.get("slice")
        if operation:
            doc.setdefault("parameters", []).append(operation["parameters"])
            doc.setdefault("responses", {}).update(operation["response"])
            success_status_codes = doc_info.get("success_status_codes", [])
            for success_status_code in success_status_codes:
                self._document_slice_metadata(
                    spec, doc["responses"][success_status_code]
                )
        return doc

With init.py

"""
The api-server Blueprint provides a REST API for managing services catalog
"""
from flask_smorest import Blueprint
from ..smorest.slice import SliceMixin

class CatalogBlueprint(SliceMixin, Blueprint):

    def __init__(self, *args,**kwargs):
        super(CatalogBlueprint,self).__init__(*args, **kwargs)

catalog_blueprint = CatalogBlueprint('catalog', __name__, template_folder='templates')

from . import routes

Then in my routes.py

    @catalog_blueprint.slice()
    def get(self, slice_parameters):
        """List all catalog nodes

        Return a list of all catalog nodes
        """
        item_count = CatalogNodeModel.query.count()
        catalog = (CatalogNodeModel.
            query.
            order_by(CatalogNodeModel.id).
            slice(
            start=getattr(slice_parameters, "start", 1),
            stop=getattr(slice_parameters, "end", item_count))
        )

        return ({"nodes": catalog.all()
                }, 200, {"X-Total-Count": item_count})

[cydergoth]

The Swagger docs aren't showing up the slice parameters....