Flask API Pagination

app.py

from flask import Flask
from flask_marshmallow import Marshmallow
from flask_sqlalchemy import SQLAlchemy

from models import Magazine
from utils import build_link_header, validate_per_page

app = Flask(__name__)
db = SQLAlchemy(app)
ma = Marshmallow(app)

app.config["SECRET_KEY"] = os.getenv("SECRET_KEY")
app.config["SQLALCHEMY_DATABASE_URI"] = os.getenv("DATABASE_URL")


@app.route("/magazines")
def magazines():
    # process query parameters
    page = request.args.get("page", 1, type=int)
    per_page = validate_per_page(request.args.get("per-page", 100, type=int))

    # query
    magazines = Magazine.query.paginate(page, per_page) 

    # map with schema
    magazine_schema = MagazineSchema()
    magazines_dumped = magazine_schema.dump(magazines.items, many=True)    

    # combined results with pagination
    results = {
        "results": magazines_dumped,
        "pagination":
            {
                "count": magazines.total,
                "page": page,
                "per_page": per_page,
                "pages": magazines.pages,
            },
    }
    
    # paginated link headers
    base_url = "https://api.mysite.org/magazines"
    link_header = build_link_header(
        query=magazines, base_url=base_url, per_page=per_page
    )
    return jsonify(results), 200, link_header


@app.errorhandler(APIError)
def handle_exception(err):
    """Return custom JSON when APIError or its children are raised"""
    # credit: https://medium.com/datasparq-technology/flask-api-exception-handling-with-custom-http-response-codes-c51a82a51a0f
    response = {"error": err.description, "message": ""}
    if len(err.args) > 0:
        response["message"] = err.args[0]
    # Add some logging so that we can monitor different types of errors
    app.logger.error("{}: {}".format(err.description, response["message"]))
    return jsonify(response), err.code

exceptions.py

class APIError(Exception):
    """All custom API Exceptions"""

    pass


class APIPaginationError(APIError):
    """Error when per-page parameter is out of bounds."""

    code = 403
    description = "pagination error"

models.py

from app import db


class Magazine(db.Model):
    __tablename__ = "magazines"

    id = db.Column(db.Integer, primary_key=True)
    publisher = db.Column(db.String(100), nullable=False)
    title = db.Column(db.String(200), nullable=False)

schemas.py

from app import ma
from models import Magazine


class MagazineSchema(ma.SQLAlchemyAutoSchema):
    class Meta:
        model = Magazine
        fields = ("title", "publisher")

utils.py

from exceptions import APIPaginationError


def build_link_header(query, base_url, per_page):
    """
    Adds pagination link headers to an API response.
    """
    links = [
        '<{0}?page=1&per-page={1}>; rel="first"'.format(base_url, per_page),
        '<{0}?page={1}&per-page={2}>; rel="last"'.format(
            base_url, query.pages, per_page
        ),
    ]
    if query.has_prev:
        links.append(
            '<{0}?page={1}&per-page={2}>; rel="prev"'.format(
                base_url, query.prev_num, per_page
            )
        )
    if query.has_next:
        links.append(
            '<{0}?page={1}&per-page={2}>; rel="next"'.format(
                base_url, query.next_num, per_page
            )
        )

    links = ",".join(links)
    return dict(Link=links)

  
def validate_per_page(per_page):
    if per_page and per_page > 100 or per_page < 1:
        raise APIPaginationError("per-page parameter must be between 1 and 100")

    return per_page